feat: 缺口3修复——sector_index_builder加工层行业指数聚合(stock_sectors_em权威映射5061只/307行业+stock_daily成分股等权聚合,统一em命名,与回测prepare_sector_context对齐)+cron 17:00

deploy/profile-scripts/sector_index_builder.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""sector_index_builder.py — 行业指数加工（数据加工层）
+
+背景（2026-08-12 架构补缺·加工层）：
+  sector_index_daily 299 行业数据来自历史回填，从未有日常采集者，8/4 停更。
+  行业指数属加工层产物——stock_sectors_em 口径的行业指数官方拉不到
+  （EM 行业分类与东财/THS 官方指数命名不一致），故由加工层聚合：
+  读 stock_sectors_em 成分股映射 + stock_daily 日K → 等权聚合行业指数。
+
+原则（老莫定）：能拉取的拉取，拉不了的才自己算。
+  stock_sectors_em(EM体系,5061只,307行业) 与官方指数命名不一致 → 自己算。
+  统一 sector=em 命名，全覆盖、不重复、与回测 prepare_sector_context 对齐。
+
+调度：收盘后跑（stock_daily 采集完成后），30 17 * * 1-5
+规范：单例守卫(5.3) + INSERT OR REPLACE 幂等 + 增量（最近N天）
+"""
+import sys, os, sqlite3, fcntl, time
+from pathlib import Path
+from datetime import datetime, timedelta
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+RECENT_DAYS = 5        # 增量：每次只算最近 5 天（历史已有，每日新增）
+
+
+def _singleton_guard(script_tag="sector_index_builder.py"):
+    """单例守卫（规范5.3）"""
+    lock_dir = Path("/tmp/mofin_locks")
+    lock_dir.mkdir(exist_ok=True)
+    try:
+        fd = os.open(str(lock_dir / f"{script_tag}.lock"), os.O_CREAT | os.O_RDWR)
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return fd
+    except OSError:
+        print(f"[{script_tag}] 已有实例在运行，退出", flush=True)
+        sys.exit(0)
+
+
+def main():
+    _fd = _singleton_guard()
+    t0 = time.time()
+    print(f"[sector_index_builder] {datetime.now().strftime('%H:%M:%S')} 行业指数加工开始", flush=True)
+
+    import pandas as pd
+    conn = sqlite3.connect(str(DB_PATH), timeout=60)
+
+    # 1. 行业映射（stock_sectors_em 权威源，5061只/307行业）
+    em = pd.read_sql("SELECT code, sector FROM stock_sectors_em", conn)
+    n_sec = em["sector"].nunique()
+    print(f"  行业映射: {len(em)} 条, {n_sec} 个行业", flush=True)
+
+    # 2. 读成分股近期日K（stock_daily，最近 RECENT_DAYS 个交易日）
+    since = (datetime.now() - timedelta(days=RECENT_DAYS * 3)).strftime("%Y-%m-%d")
+    klines = pd.read_sql(
+        "SELECT code, date, close, high, low, volume FROM stock_daily WHERE date >= ?",
+        conn, params=[since])
+    # 只保留最近 RECENT_DAYS 个交易日
+    dates = sorted(klines["date"].unique())[-RECENT_DAYS:]
+    klines = klines[klines["date"].isin(dates)]
+    print(f"  日K: {len(klines)} 行, {len(dates)} 个交易日 ({dates[0]}~{dates[-1]})", flush=True)
+
+    # 3. 关联行业映射 → 行业指数聚合（等权均值 + 成交量求和）
+    df = klines.merge(em, on="code", how="inner")
+    print(f"  关联后: {len(df)} 行（{df['code'].nunique()} 只有行业归属）", flush=True)
+
+    grp = df.groupby(["sector", "date"], as_index=False).agg(
+        close=("close", "mean"),
+        high=("high", "mean"),
+        low=("low", "mean"),
+        volume=("volume", "sum"),
+    )
+    # 行业日涨跌：按 sector 排序后相邻日期收盘均值变化
+    grp = grp.sort_values(["sector", "date"])
+    grp["prev_close"] = grp.groupby("sector")["close"].shift(1)
+    grp["change_pct"] = ((grp["close"] / grp["prev_close"]) - 1) * 100
+    grp = grp.drop(columns=["prev_close"])
+
+    print(f"  行业指数: {len(grp)} 行（{grp['sector'].nunique()} 行业 × {len(dates)} 天）", flush=True)
+
+    # 4. 写 sector_index_daily（INSERT OR REPLACE 幂等）
+    cur = conn.cursor()
+    written = 0
+    for r in grp.itertuples():
+        cur.execute(
+            "INSERT OR REPLACE INTO sector_index_daily (sector, date, close, change_pct, high, low) "
+            "VALUES (?,?,?,?,?,?)",
+            (r.sector, r.date, round(r.close, 3),
+             round(r.change_pct, 2) if pd.notna(r.change_pct) else None,
+             round(r.high, 3), round(r.low, 3)))
+        written += 1
+    conn.commit()
+    conn.close()
+
+    dt = time.time() - t0
+    print(f"[sector_index_builder] 完成: {written} 行写入, {grp['sector'].nunique()} 行业, 耗时 {dt:.0f}s", flush=True)
+
+
+if __name__ == "__main__":
+    main()