feat: 缺口1修复——全市场日K采集daily_kline_collector(4013只,sh/sz分批,0.05s间隔防封,600s护栏)+2个cron(16:10sh/16:25sz)

This commit is contained in:
hmo
2026-08-12 08:32:23 +08:00
parent d0b9dc8c85
commit de127d229d
3 changed files with 125 additions and 0 deletions
@@ -0,0 +1,115 @@
#!/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 维重评/策略扫描直接读本地。
数据源:腾讯前复权日Kmarket_data.fetch_tx_klines,与 stock_daily 零偏差)
调度:分 2 个 cronsh/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
规范:单例守卫(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
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")])
else:
# 0/3 开头 → 深圳
return sorted([c for c in codes if c[0] in ("0", "3")])
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
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)
codes = get_codes(conn, prefix)
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:
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()
+5
View File
@@ -0,0 +1,5 @@
#!/usr/bin/env python3
"""daily_kline_sh.py — 全市场日K采集(上海,包装入口)"""
import sys, runpy
sys.argv = ["daily_kline_collector.py", "--prefix", "sh"]
runpy.run_path("/home/hmo/.hermes/profiles/position-analyst/scripts/daily_kline_collector.py", run_name="__main__")
+5
View File
@@ -0,0 +1,5 @@
#!/usr/bin/env python3
"""daily_kline_sz.py — 全市场日K采集(深圳,包装入口)"""
import sys, runpy
sys.argv = ["daily_kline_collector.py", "--prefix", "sz"]
runpy.run_path("/home/hmo/.hermes/profiles/position-analyst/scripts/daily_kline_collector.py", run_name="__main__")