178 lines
7.5 KiB
Python
178 lines
7.5 KiB
Python
#!/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()
|