#!/usr/bin/env python3
"""daily_kline_collector.py — 全市场日K采集（stock_daily 日常刷新）

背景（2026-08-12 架构补缺）：
  stock_daily 全市场 4272 只数据来自历史回填，日常无采集 cron——
  仅 refresh_mtf_cache 顺带写持仓+自选+指数(122只)，7月起全市场数据断层。
  本脚本补齐：收盘后全市场日K刷新，供 12 维重评/策略扫描直接读本地。

数据源：腾讯前复权日K（market_data.fetch_tx_klines，与 stock_daily 零偏差）
调度：分 2 个 cron（sh/sz 分批，600s 护栏内，0.05s 间隔避免被封）
  sh: 10 16 * * 1-5  → daily_kline_collector.py --prefix sh
  sz: 25 16 * * 1-5  → daily_kline_collector.py --prefix sz

港股批（阶段2B 新增，2026-08-14）：
  python3 daily_kline_collector.py hk   （或 --prefix hk）
  股票源 = get_stock_pool(market='hk')（hk_connect_stocks 名单 + 港股持仓/自选），
  fetch_tx_klines 经 kline_symbol 自动加 hk 前缀（hk00700），接口已兼容港股
  （qfqday/day 双键）。hk_connect_stocks 表不存在时优雅降级：打印提示并跳过。

规范：单例守卫(5.3) + INSERT OR REPLACE 幂等 + 限速防封 + 分批 commit
"""
import sys, os, json, time, sqlite3, fcntl
from pathlib import Path
from datetime import datetime

sys.path.insert(0, str(Path(__file__).parent))
from market_data import fetch_tx_klines, get_stock_pool

DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
SLEEP = 0.05          # 请求间隔（≈5 请求/秒，避免被腾讯封）
DATALEN = 75          # 拉 75 根（>70 满足 fetch_tx_klines 下限），只写最新 5 根
WRITE_RECENT = 5      # 只写最新 5 根（历史已有，增量刷新）
BATCH_COMMIT = 50     # 每 50 只 commit 一次


def _singleton_guard(script_tag):
    """单例守卫（规范5.3）：防重复实例并发写"""
    lock_dir = Path("/tmp/mofin_locks")
    lock_dir.mkdir(exist_ok=True)
    lock_path = lock_dir / f"{script_tag}.lock"
    try:
        fd = os.open(str(lock_path), 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 get_codes(conn, prefix):
    """全市场 A 股代码（stock_daily distinct，与回测/扫描同口径）"""
    rows = conn.execute(
        "SELECT DISTINCT code FROM stock_daily WHERE length(code)=6"
    ).fetchall()
    codes = [str(r[0]) for r in rows]
    if prefix == "sh":
        # 6/9/5 开头 → 上海（fetch_tx_klines 自动加 sh 前缀）
        return sorted([c for c in codes if c[0] in ("5", "6", "9")])
    elif prefix == "sz_a":
        # 深圳前半批（0 开头）——2026-08-13 分批：sz 2180只超时，分 sz_a/sz_b
        return sorted([c for c in codes if c[0] == "0"])
    elif prefix == "sz_b":
        # 深圳后半批（3 开头）
        return sorted([c for c in codes if c[0] == "3"])
    else:
        # 0/3 开头 → 深圳（全部，兼容旧调用）
        return sorted([c for c in codes if c[0] in ("0", "3")])


def get_hk_codes():
    """港股代码：get_stock_pool(market='hk')（hk_connect_stocks 名单 + 港股持仓/自选）

    hk_connect_stocks 表不存在（阶段2A 并行开发中）→ 优雅降级：
    打印提示后跳过港股批，不报错；表存在但港股池为空同样跳过。
    """
    # 先探测表是否存在，给精确的降级提示（get_stock_pool 内部也容错，双保险）
    try:
        with sqlite3.connect(str(DB_PATH), timeout=5) as conn:
            conn.execute("SELECT 1 FROM hk_connect_stocks LIMIT 1").fetchone()
    except sqlite3.OperationalError as e:
        if "no such table" in str(e).lower():
            print("  hk_connect_stocks 表不存在，跳过港股批", flush=True)
            return []
        raise
    codes, _ = get_stock_pool(market="hk")
    codes = sorted(codes)
    if not codes:
        print("  港股池为空，跳过港股批", flush=True)
        return []
    return codes


def main():
    prefix = "sz"
    limit = 0
    i = 1
    while i < len(sys.argv):
        a = sys.argv[i]
        if a.startswith("--prefix"):
            prefix = a.split("=")[-1] if "=" in a else sys.argv[i + 1]
        if a.startswith("--limit"):
            limit = int(a.split("=")[-1] if "=" in a else sys.argv[i + 1])
        i += 1
    # 位置子命令：python3 daily_kline_collector.py hk（与 --prefix hk 等价）
    if len(sys.argv) > 1 and sys.argv[1] == "hk":
        prefix = "hk"
    tag = f"daily_kline_collector_{prefix}"
    _fd = _singleton_guard(tag)

    t0 = time.time()
    print(f"[{tag}] {datetime.now().strftime('%H:%M:%S')} 全市场日K采集({prefix}) 开始", flush=True)

    conn = sqlite3.connect(str(DB_PATH), timeout=30)
    if prefix == "hk":
        codes = get_hk_codes()          # 港股池：hk_connect_stocks + 港股持仓/自选
    else:
        codes = get_codes(conn, prefix) # A股池：stock_daily distinct 分批
    if limit > 0:
        codes = codes[:limit]
        print(f"  [测试模式] 只跑前 {limit} 只", flush=True)
    print(f"  股票池: {len(codes)} 只", flush=True)

    cur = conn.cursor()
    ok = fail = empty = written = 0
    for i, code in enumerate(codes, 1):
        try:
            bars = fetch_tx_klines(code, datalen=DATALEN)
            if not bars:
                empty += 1
                continue
            # 只写最新 WRITE_RECENT 根（历史已有，增量刷新，减少写入）
            for b in bars[-WRITE_RECENT:]:
                cur.execute(
                    "INSERT OR REPLACE INTO stock_daily (code, date, open, close, high, low, volume) "
                    "VALUES (?,?,?,?,?,?,?)",
                    (code, b["date"], b["open"], b["close"], b["high"], b["low"], b["volume"]),
                )
                written += 1
            ok += 1
        except Exception as e:
            # 2026-08-18 database is locked 指数退避重试（工单t_17ba6f5d：sz_b整点调度撞锁致2只失败）
            if "database is locked" in str(e).lower() or "locked" in str(e).lower():
                import random as _rd
                _retry = 0
                _ok = False
                while _retry < 3 and not _ok:
                    try:
                        time.sleep(0.5 * (2 ** _retry) + _rd.uniform(0, 0.3))
                        for b in bars[-WRITE_RECENT:]:
                            cur.execute(
                                "INSERT OR REPLACE INTO stock_daily (code, date, open, close, high, low, volume) "
                                "VALUES (?,?,?,?,?,?,?)",
                                (code, b["date"], b["open"], b["close"], b["high"], b["low"], b["volume"]),
                            )
                        conn.commit()
                        _ok = True
                    except Exception:
                        _retry += 1
                if _ok:
                    ok += 1
                    continue
            fail += 1
            if fail <= 5:
                print(f"  FAIL {code}: {e}", flush=True)
        if i % BATCH_COMMIT == 0:
            conn.commit()
            print(f"  [{i}/{len(codes)}] ok={ok} empty={empty} fail={fail} written={written} | {time.time()-t0:.0f}s", flush=True)
        time.sleep(SLEEP)

    conn.commit()
    conn.close()
    dt = time.time() - t0
    print(f"[{tag}] 完成: {ok}/{len(codes)} 成功, {empty} 空, {fail} 失败, 写入 {written} 根, 耗时 {dt:.0f}s", flush=True)


if __name__ == "__main__":
    main()