diff --git a/deploy/profile-scripts/sector_index_builder.py b/deploy/profile-scripts/sector_index_builder.py new file mode 100644 index 00000000..a83fea91 --- /dev/null +++ b/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()