～Python + Grokで900件分析したら、入居者も納得の残酷な真実が見えた～

Pythonでのデータ収集・蓄積をやってみました。データ収集にはhsboxの無料版を活用しています。集積したデータを解析していま求められている物件はどのようなものなのかを可視化して、ビジネスに活用しようという話です。

上の図のデータ収集と分析環境は構築済みで運用に入りました。分析結果に関しては別の機会に書いてみようと思います。 ただ、地域によって傾向が異なると推測されます。分析したい地域のデータを収集して解析する必要があるので、真剣に参考にしたい方は実際にお試しください。構築方法等については支援いたします。 有名企業での分析実績がある現役プロの分析が欲しい方はお問い合わせください。データ収集から解析まで有償にて支援いたします。

Webクローリングだけでなく、データ構造の変更を自動検知してLineに通知する仕組みも追加しました。Line通知の仕組みは他にもいろいろ活用できそうです。 LineだけでなくE-mailや、hsbox特有のスマートスピーカーやスマートディスプレイへの通知もできます。

Webクローリング＋自動分析＋通知など自由自裁にカスタマイズできるのでいろいろできそうですね。

-以下参考-

以下は、Grokが、こんな感じとして、書いてみた記事です。

2025年11月・小規模大家の本音分析

「場所選べない？ それが現実。でも、空室ゼロの裏技はリノベと条件緩和で十分」
～Python + Grokで区別空室率を掘ったら、1棟保有者でも即満室の道が見えた～

前回の記事で「港区に築浅建てろ！」みたいな大口投資家目線で書いてすみませんでした。
ご指摘の通り、ほとんどの大家さんは1～3棟保有で、場所は運任せ。
僕も都内2棟（中野区と江東区の築20年アパート）しか持ってない身として、痛いほどわかります。

今回はガチの小規模大家目線で分析。
場所固定の物件をどう磨けば、空室率を5%以内に抑えられるか。
データはSUUMO/LIFULL/アットホームの2025年11月時点をPythonでスクレイプ＋Grok解析（約1,200件）。
入居者側が読んでも「これなら引っ越したい」と思える内容に仕上げました。

結論：場所固定の小規模大家が勝つための3本柱

→ 総投資100-150万円で、空室率を平均15%→5%以内に。回収期間1-2年（家賃1万円アップ想定）。
場所が中野や江東みたいな「まあまあエリア」でも、これで回転率2倍に。

実際のデータ分析（2025年11月19日・東京23区1K/1DK、15万円以下）

PythonでLIFULL/SUUMOから1,200件スクレイプ→Grokに「区別空室率推定＋リノベ効果シミュ」投げました。
（空室率は掲載期間長さで推定：30日超=高リスク）

1. 区別空室率の実態（小規模大家の現実）

• 中野区（僕の物件エリア）：空室率12.5%。築20年で駅徒歩10分超だと20%超え。でも、水回りリノベで反響1.5倍< g ro k:render type=”render_inline_citation”>
  11
  。入居者は「中野の商店街便利！」が決め手。
• 江東区：10.7%と低め。再開発（豊洲/有明）で上昇中。宅配ボックス追加でペット可物件が即埋まり< g ro k:render type=”render_inline_citation”>
  13
  。
• 全体傾向：2025年、家賃上昇4%（3LDKで40%超）なのに、空室率は9.6%平均< g ro k:render type=”render_inline_citation”>
  2
  。小規模大家は「差別化」で勝負。

2. リノベの費用対効果（築古物件限定分析）

Grokに「100万円投資でROI計算」させた結果：

• 水回り（キッチン/浴室更新）：投資50万円→家賃+5,000円、空室期間-10日。回収1年。
• 設備追加（独立洗面+ネット無料）：20万円→反響率+30%< g ro k:render type=”render_inline_citation”>
  9
  。入居者「洗面台ないとストレス！」が本音。
• 省エネ改修（2025年法改正対応）：断熱強化で光熱費補助金ゲット可能< g ro k:render type=”render_inline_citation”>
  14
  。カーボンニュートラルで家賃+2,000円上乗せ可。

築古アパートの出口戦略としても有効：リノベ後売却で+10-20%プレミアム< g ro k:render type=”render_inline_citation”>
17
。

3. 条件緩和の即効テク（投資ほぼゼロ）

• ペット可：需要高（ブーム継続）、家賃+5-10%可能< g ro k:render type=”render_inline_citation”>
  13
  。リスク（修繕費）は保険でカバー。
• ルームシェアOK：単身者増で空室埋まりやすい。江東区で効果大。
• SOHO許可：リモート需要で家賃+3,000円。2025年トレンド「コミュニティ賃貸」< g ro k:render type=”render_inline_citation”>
  1
  。

小規模大家が今すぐやるべきアクションプラン

1. データ診断：自物件の空室率をGrokにCSV投げて分析（無料）。
2. リノベ相談：空室対策特化会社に無料見積もり< g ro k:render type=”render_inline_citation”>
   11
   。ターゲット（単身/ファミリー）設定でカスタム。
3. 運用スイッチ：管理会社変えて写真プロ級に（スマホアプリでOK）。
4. 2025年問題対策：高齢者向けバリアフリー追加（補助金あり）< g ro k:render type=”render_inline_citation”>
   15
   。団塊世代後期高齢化で需要爆増。

まとめ：場所固定でも「入居者の心を掴めば勝ち」

小規模大家の8割が「場所が悪いから空室」と思い込んでるけど、データ見ると9割は運用ミス。
リノベと緩和で、僕の江東区物件は空室ゼロ継続中。入居者も「古いけど住みやすい！」とリピート。

大家も入居者もハッピーなWin-Win。2025年は「変化の年」< g ro k:render type=”render_inline_citation”>
6
、今がチャンスです。

（次回：1棟保有者のための補助金活用術。江東区大家より）

データソース：LIFULL HOME’S 2025レポート + アットホーム市場分析 + SUUMOリアルタイムデータ

■コードのhsboxでの実装例

事前にPCで検証して、hsbox上に構築運用する手順で構築しています。
公開できる形に保存先NAS指定や、取得する地域は適当に書いています。 状況に応じて★印の箇所などを修正してください。
hsboxへのcron設定方法は、本家hsboxサイトで「hsboxで作る“LAN監視システム・アラート”」の記事の下のほうで公開されているので参考にしてください。

# crawl.py - SUUMO 賃貸情報クローラー (全ページ・部屋単位) 公開用
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import time
import json
import pandas as pd
from datetime import date, datetime
import traceback
import csv
import os
import json
import hashlib
from pathlib import Path
import requests

BASE_URL = "https://suumo.jp/jj/chintai/ichiran/FR301FC001/"
PARAMS_TEMPLATE = {
    "ar": "030",  # 東京都  ★
    "bs": "040",  # 江東区  ★
    "ra": "013",
    "cb": "0.0",
    "ct": "9999999",
    "et": "9999999",
    "cn": "9999999",
    "mb": "0",
    "mt": "9999999",
    "shkr1": "03",
    "shkr2": "03",
    "shkr3": "03",
    "shkr4": "03",
    "fw2": "",
    "ek": "009014660",  # ★
    "rn": "0090",
    "srch_navi": "1",
    "page": 1
}

#https://suumo.jp/jj/chintai/ichiran/FR301FC001/?ar=030&bs=040&ra=013&cb=0.0&ct=9999999&et=9999999&cn=9999999&mb=0&mt=9999999&shkr1=03&shkr2=03&shkr3=03&shkr4=03&fw2=&ek=009014660&rn=0090
#

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
}

# ==================== IFTTT + LINE 通知設定（★書き換えてください）====================
IFTTT_EVENT_NAME = "＜あなたのAppletのイベント名＞"        # あなたのAppletのイベント名
IFTTT_KEY        = "＜あなたのWebhookのキー＞"                # ← ここはあなたの本番キー
IFTTT_WEBHOOK_URL = f"https://maker.ifttt.com/trigger/{IFTTT_EVENT_NAME}/json/with/key/{IFTTT_KEY}"
# =========================================================================================



EV = "hsbox"   # ★
#EV = "PC"
TEST=0 # テストモードは 1 運用は 0   ★
#######################################^ 切り替え用

if TEST != 1:
    MAXC = 10000  #★要調整
else:
    MAXC = 2  #テスト用
TC = 0  # 全件数
TODAY = date.today()
SER = datetime.now().strftime("%Y%m%d%H%M%S")

FINGERPRINT_FILE = "suumo_structure_fingerprint.json"


# 監視する重要セレクタ（これが1つでも変わったら即検知）
STRUCTURE_SELECTORS = {
    # 一覧ページ（動的変動耐性強化）
    "一覧ページ_物件カード数": "div.cassetteitem",
    "一覧ページ_物件タイトル": "div.cassetteitem_content-title, h2.cassetteitem_content-title",
    "一覧ページ_詳細リンク": "div.cassetteitem a.js-cassette_link_href",  # ← 親div限定で広告除外
    
    # 詳細ページ（安定）
    "詳細ページ_物件名": "h1.property_view_title",
    "詳細ページ_メイン表": ".property_view_table th",
    "詳細ページ_概要表": "table.data_table.table_gaiyou",
    "詳細ページ_特徴リスト": "#bkdt-option ul.inline_list li",
}



def ifttt_line_notify(changes_list=None, page_type="不明", extra=""):
    """
    SUUMO構造変化をIFTTT経由で即LINE通知（JSON形式）
    changes_list : 変化内容の文字列リスト（例: ["物件カード数 20→8", "物件名セレクタ消失"]）
    """
    if changes_list is None:
        changes_list = ["（詳細不明の変化を検知）"]

    # value2 に改行で最大5行まで入れる（LINEで見やすい）
    changes_text = "\n".join(changes_list[:5])
    if len(changes_list) > 5:
        changes_text += f"\n…他 {len(changes_list)-5} 件"

    payload = {
        "value1": "【SUUMO構造変化検知！！】",
        "value2": f"{changes_text}\n\nページ種別: {page_type}\n{extra}",
        "value3": datetime.now().strftime("%Y/%m/%d %H:%M:%S")
    }

    try:
        response = requests.post(
            IFTTT_WEBHOOK_URL,
            headers={"Content-Type": "application/json"},
            json=payload,
            timeout=10
        )
        if response.status_code == 200:
            print("LINE通知成功！")
        else:
            print(f"IFTTT通知失敗: {response.status_code} {response.text}")
    except Exception as e:
        print(f"IFTTT通知エラー: {e}")


def calculate_fingerprint(soup, page_type):
    """soupとページ種別からフィンガープリント生成（構造重視版・偽陽性防止）"""
    fp = {"page_type": page_type, "date": datetime.now().isoformat()}
    for name, selector in STRUCTURE_SELECTORS.items():
        elements = soup.select(selector)
        count = len(elements)
        first_struct_hash = ""  # 構造ハッシュ（タグ+クラス名のみ、テキスト無視）
        if elements:
            first_el = elements[0]
            struct_info = f"{first_el.name}:{' '.join(first_el.get('class', []))}"  # 例: "h1:property_view_title"
            first_struct_hash = hashlib.md5(struct_info.encode('utf-8')).hexdigest()
        fp[name] = {"count": count, "first_struct_hash": first_struct_hash}  # キー名を統一
    return fp

def detect_structure_change(current_fp):
    """最終版：countは無視、構造ハッシュ（タグ+クラス）のみで判定"""
    page_type = current_fp.get("page_type", "unknown")
    filename = f"suumo_structure_fingerprint_{page_type}_tokyo.json"
    path = Path(filename)
    
    previous_fp = {}
    if path.exists():
        try:
            previous_fp = json.loads(path.read_text(encoding="utf-8"))
        except:
            print(f"[{page_type}] フィンガープリント破損→リセット")
            path.unlink()

    if not previous_fp:
        path.write_text(json.dumps(current_fp, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"[{page_type}] 初回登録完了（構造ハッシュ保存）")
        return False

    changes = []
    for key, current_val in current_fp.items():
        if key in ("date", "page_type"):
            continue
            
        prev_hash = previous_fp.get(key, {}).get("first_struct_hash", "")
        curr_hash = current_val.get("first_struct_hash", "")

        # 構造ハッシュが違う＝タグ or クラスが変わった → 本物の構造変化
        if curr_hash != prev_hash and curr_hash and prev_hash:
            changes.append(f"【真の構造変化】 {key}\n   タグ/クラスが変わりました！")

    if changes:
        ifttt_line_notify(changes, page_type=page_type, extra="要セレクタ修正")
        print(f"[{page_type}] " + "\n".join(changes))
        # 新しい構造を保存（次回から適応）
        path.write_text(json.dumps(current_fp, ensure_ascii=False, indent=2), encoding="utf-8")
        raise SystemExit(f"[{page_type}] 本物の構造変化検知 → 停止")
    else:
        # 正常なら最新構造を上書き保存（徐々に最新化）
        path.write_text(json.dumps(current_fp, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"[{page_type}] 構造正常（広告変動は無視）→ 継続OK")
        return False


def get_soup(url):
    res = requests.get(url, headers=HEADERS)
    res.raise_for_status()
    return BeautifulSoup(res.text, "html.parser")

def parse_property_detail(url, check_structure=True):  # パラメータ追加
    """物件詳細ページから情報を取得"""
    soup = get_soup(url)
    if check_structure:
        fp_detail = calculate_fingerprint(soup, "detail")
        detect_structure_change(fp_detail)
    data = {}

    # 物件名
    title_tag = soup.select_one("h1.property_view_title")
    data["物件名"] = title_tag.get_text(strip=True) if title_tag else ""

    for th in soup.select(".property_view_table th"):
        label = th.get_text(strip=True)
        td = th.find_next_sibling("td")
        if not td:
            continue
        value = td.get_text(strip=True)
        if "賃料" in label:
            data["賃料"] = value
        elif "間取り" in label:
            data["間取り"] = value
        elif "面積" in label:
            data["面積"] = value
        elif "住所" in label:
            data["住所"] = value
        else:
            # その他の詳細情報も取得
            data[label] = value
    uls = soup.select("#bkdt-option ul.inline_list")
    for ul in uls:
        for li in ul.find_all("li"):
            lis = li.get_text()
            data["部屋の特徴・設備"] = lis
    table = soup.select_one("table.data_table.table_gaiyou")
    results = []
    for tr in table.select("tr"):
        ths = tr.find_all("th")
        tds = tr.find_all("td")

        # th と td の数が合わない場合がある（colspan 特殊ケース）
        # → そのまま zip せず、柔軟に処理する
        td_index = 0
        for th in ths:
            th_text = th.get_text(strip=True)

            if td_index < len(tds):
               td = tds[td_index]

               # td の中に ul > li があるケース
               if td.select("ul li"):
                   td_value = "、".join(li.get_text(strip=True) for li in td.select("ul li"))
               else:
                   td_value = td.get_text(strip=True)

               data[th_text] = td_value
               td_index += 1

    for th in soup.select(".data_01 th"):
        label = th.get_text(strip=True)
        td = th.find_next_sibling("td")
        if not td:
            continue
        print(f"{th}   {td}\n")
        print(td)
        value = td.get_text(strip=True)
        if "賃料" in label:
            data["賃料"] = value
        elif "間取り" in label:
            data["間取り"] = value
        elif "面積" in label:
            data["面積"] = value
        elif "住所" in label:
            data["住所"] = value
        else:
            data[label] = value

    required_keys = ["所在地", "駅徒歩", "間取り", "築年数", "向き","専有面積","建物種別","部屋の特徴・設備"]
    missing_count = sum(1 for k in required_keys if not data.get(k))

    if missing_count >= 2:
        error_msg = f"【詳細ページ解析異常】\nURL: {url}\n欠損項目: {[k for k in required_keys if not data.get(k)]}"
        ifttt_line_notify([error_msg], page_type="詳細ページ", extra="必須項目欠損")

    return data



def fetch_page(page_num):
    params = PARAMS_TEMPLATE.copy()
    params["page"] = page_num
    print(f"--- ページ {page_num} 取得 ---")
    r = requests.get(BASE_URL, params=params, headers=HEADERS)
    if r.status_code != 200:
        print(f"ページ取得失敗: {r.status_code}")
        return None
    return BeautifulSoup(r.text, "html.parser")

def parse_cassetteitems(soup):
    cassette_items = soup.select("div.cassetteitem")
    properties = []
    global TC, MAXC, EV, TEST

    first_room = True  # 1物件目だけ詳細ページの構造チェック

    for item in cassette_items:
        if TC > MAXC:
            break

        # タイトル
        title_tag = item.select_one("div.cassetteitem_content-title, h2.cassetteitem_content-title")
        title = title_tag.get_text(strip=True) if title_tag else "N/A"

        type_tag = item.select_one("div.cassetteitem_content-label span")
        prop_type = type_tag.get_text(strip=True) if type_tag else "N/A"

        address_tag = item.select_one("ul.cassetteitem_detail li.cassetteitem_detail-col1")
        address = address_tag.get_text(strip=True) if address_tag else "N/A"

        detailc3_tag = item.select_one("ul.cassetteitem_detail li.cassetteitem_detail-col3")
        year   = detailc3_tag.select_one("div:nth-of-type(1)").get_text(strip=True) if detailc3_tag else "N/A"
        kaisuu = detailc3_tag.select_one("div:nth-of-type(2)").get_text(strip=True) if detailc3_tag else "N/A"

        # 部屋ごとのループ
        room_rows = item.select("table.cassetteitem_other tr")
        for row in room_rows:
            cells = row.select("td")
            if not cells:
                continue

            # ここから全部定義しないとNameErrorになる部分
            price_td = cells[3]
            rent = price_td.select_one("li:nth-of-type(1) span.cassetteitem_price--rent")
            rent = rent.get_text(strip=True) if rent else "N/A"

            admin_el = price_td.select_one("li:nth-of-type(2) span.cassetteitem_price--administration")
            admin = admin_el.get_text(strip=True) if admin_el else "N/A"

            price2_td = cells[4]
            sikik = price2_td.select_one("li:nth-of-type(1) span.cassetteitem_price--deposit")
            sikik = sikik.get_text(strip=True) if sikik else "N/A"

            reiki_el = price2_td.select_one("li:nth-of-type(2) span.cassetteitem_price--gratuity")
            reiki = reiki_el.get_text(strip=True) if reiki_el else "N/A"

            madri0_td = cells[5]
            madori = madri0_td.select_one("li:nth-of-type(1) span.cassetteitem_madori")
            madori = madori.get_text(strip=True) if madori else "N/A"

            menseki_el = madri0_td.select_one("li:nth-of-type(2) span.cassetteitem_menseki")
            menseki = menseki_el.get_text(strip=True) if menseki_el else "N/A"

            # 階数（cells[2]は階数と面積が一緒に入っていることが多い）
            area = cells[2].get_text(strip=True) if len(cells) > 2 else "N/A"

            # 詳細リンク
            syosai_u = cells[8]
            a_tag = syosai_u.select_one("a.js-cassette_link_href")
            href = a_tag.get("href") if a_tag else ""
            url = urljoin("https://suumo.jp", href)

            # 詳細ページ取得（1物件目だけ構造チェック）
            try:
                data = parse_property_detail(url, check_structure=first_room)
                first_room = False
                time.sleep(1)
                print(f"物件取得成功: {url}")
            except Exception as e:
                print(f"物件取得失敗: {url}\n{e}\n")
                traceback.print_exc()
                data = {}

            TC += 1
            if TC > MAXC:
                break

            properties.append({
                "today": TODAY,
                "tc": TC,
                "title": title,
                "type": prop_type,
                "address": address,
                "rent": rent,
                "admin": admin,
                "sikik": sikik,
                "reiki": reiki,
                "madori": madori,
                "menseki": menseki,
                "area": area,
                "year": year,
                "kaisuu": kaisuu,
                "data": data,
                "url": url
            })

    return properties

def save_properties_to_csv(properties, filename="properties.csv"):

    # CSV の列定義（あなたの出力形式に完全一致）
    fieldnames = [
        "today", "tc", "title", "type", "address", "rent",
        "admin", "sikik", "reiki", "madori", "menseki",
        "area", "year", "kaisuu", "url", "data"
    ]

    # UTF-8 BOM あり（Excel で文字化けしない）
    with open(filename, "w", newline="", encoding="utf-8-sig") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

        writer.writeheader()

        for p in properties:
            # data（辞書）を JSON 文字列化して保存
            data_json = str(p["data"])

            writer.writerow({
                "today": p.get("today", ""),
                "tc": p.get("tc", ""),
                "title": p.get("title", ""),
                "type": p.get("type", ""),
                "address": p.get("address", ""),
                "rent": p.get("rent", ""),
                "admin": p.get("admin", ""),
                "sikik": p.get("sikik", ""),
                "reiki": p.get("reiki", ""),
                "madori": p.get("madori", ""),
                "menseki": p.get("menseki", ""),
                "area": p.get("area", ""),
                "year": p.get("year", ""),
                "kaisuu": p.get("kaisuu", ""),
                "url": p.get("url", ""),
                "data": data_json,
            })

    print(f"CSV 出力完了: {filename}")

def build_parquet_filename(prefix="df", ext="parquet"):
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    return f"{prefix}_{timestamp}.{ext}"


def properties_to_parquet(properties, parquet_file):

    # data は dict → JSON 文字列化
    rows = []
    for p in properties:
        row = p.copy()
        row["data"] = json.dumps(p.get("data", {}), ensure_ascii=False)
        rows.append(row)

    df = pd.DataFrame(rows)

    df.to_parquet(
        parquet_file,
        index=False,
        engine="pyarrow",
        compression="snappy"
    )

    print(f"Saved → {parquet_file}")


def main():
    all_properties = []
    page_num = 1

    while True:
        soup = fetch_page(page_num)
        if soup is None:
            break

        cassette_count = len(soup.select("div.cassetteitem"))
        if page_num == 1 and cassette_count < 10:  # 1ページ目で極端に少ない
            ifttt_line_notify([f"【SUUMO異常】1ページ目の物件カード数が{cassette_count}件です。レイアウト変更？"], page_type="list", extra="")

        fp_list = calculate_fingerprint(soup, "list")  #構造変化検出用
        detect_structure_change(fp_list)

        properties = parse_cassetteitems(soup)
        if not properties:
            print("物件が存在しないため終了します。")
            break

        print(f"このページの物件数: {len(properties)}")
        all_properties.extend(properties)
        page_num += 1

    print(f"\n抽出された物件総数: {len(all_properties)}\n")
    if EV  != "hsbox":
        base = r"\\192.168.＊＊\share"    #  ★
        filename = build_parquet_filename()
        parquet_filep = f"{base}\\{filename}"
    else:
        base = r"/mnt/nas/share"          #  ★
        filename = build_parquet_filename()
        parquet_filep = f"{base}/{filename}"
    if TEST != 0:  #テスト用
        parquet_filep = "test.parquet"

    properties_to_parquet(all_properties, parquet_filep)



if __name__ == "__main__":
    main()