#!/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()