106 lines
4.3 KiB
Python
106 lines
4.3 KiB
Python
#!/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
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
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()
|