From 29db5efd9568fe65e94118955145a00e3c52f319 Mon Sep 17 00:00:00 2001 From: hmo Date: Fri, 14 Aug 2026 16:39:17 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B8=AF=E8=82=A1=E6=8E=A5=E5=85=A5?= =?UTF-8?q?=E9=98=B6=E6=AE=B52=E2=80=94=E2=80=94hk=5Fconnect=5Flist?= =?UTF-8?q?=E6=B8=AF=E8=82=A1=E9=80=9A=E5=90=8D=E5=8D=95=E9=87=87=E9=9B=86?= =?UTF-8?q?=E5=99=A8(=E4=B8=9C=E8=B4=A2MK0146=E6=9D=BF=E5=9D=97620?= =?UTF-8?q?=E5=8F=AA,push2delay=E9=95=9C=E5=83=8F=E9=81=BF=E5=BC=80?= =?UTF-8?q?=E4=B8=BBhost=E5=B0=81=E7=A6=81)+get=5Fstock=5Fpool(market=3D)?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E5=8C=96(=E9=BB=98=E8=AE=A4A=E8=82=A1?= =?UTF-8?q?=E8=A1=8C=E4=B8=BA=E4=B8=8D=E5=8F=98)+daily=5Fkline=5Fcollector?= =?UTF-8?q?=E6=B8=AF=E8=82=A1=E6=89=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../profile-scripts/daily_kline_collector.py | 39 ++- deploy/profile-scripts/hk_connect_list.py | 252 ++++++++++++++++++ deploy/profile-scripts/market_data.py | 43 ++- 3 files changed, 325 insertions(+), 9 deletions(-) create mode 100644 deploy/profile-scripts/hk_connect_list.py diff --git a/deploy/profile-scripts/daily_kline_collector.py b/deploy/profile-scripts/daily_kline_collector.py index 72348b89..40e8e6d6 100644 --- a/deploy/profile-scripts/daily_kline_collector.py +++ b/deploy/profile-scripts/daily_kline_collector.py @@ -11,6 +11,12 @@ 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 @@ -18,7 +24,7 @@ from pathlib import Path from datetime import datetime sys.path.insert(0, str(Path(__file__).parent)) -from market_data import fetch_tx_klines +from market_data import fetch_tx_klines, get_stock_pool DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") SLEEP = 0.05 # 请求间隔（≈5 请求/秒，避免被腾讯封） @@ -61,6 +67,29 @@ def get_codes(conn, prefix): 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 @@ -72,6 +101,9 @@ def main(): 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) @@ -79,7 +111,10 @@ def main(): 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 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) diff --git a/deploy/profile-scripts/hk_connect_list.py b/deploy/profile-scripts/hk_connect_list.py new file mode 100644 index 00000000..9a3fc373 --- /dev/null +++ b/deploy/profile-scripts/hk_connect_list.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +"""hk_connect_list.py — 港股通标的名单采集器（东财 MK0146 板块） + +用途： + 采集港股通（沪港通+深港通）标的名单，写入 hk_connect_stocks 表。 + 老莫是港股通用户，选股范围 = 港股通标的。 + 本名单后续驱动：港股日K采集（daily_kline_collector 港股批）、 + 选股池（get_stock_pool(market='hk') 读此表）。 + 遵循系统铁律：数据采集层定期采集入本地 DB，使用层只读不采集。 + +数据源（2026-08-14 在 246 实测可用）： + 东财行情延时镜像 push2delay.eastmoney.com（主 host push2 对 246 被封： + RemoteDisconnected，必须用 push2delay 延时镜像）。 + fs=b:MK0146 = 港股通板块，total=620 只；单页上限 100 条，翻 7 页（pn=1..7）。 + fields=f12(5位港股代码),f14(名称)；必须带 User-Agent: Mozilla/5.0。 + +表结构（严格遵守，并行任务按此读取）： + CREATE TABLE IF NOT EXISTS hk_connect_stocks ( + code TEXT PRIMARY KEY, -- 5位港股代码如 00700 + name TEXT, -- 名称 + in_date TEXT, -- 首次进入名单日期 YYYY-MM-DD + out_date TEXT, -- 调出日期（NULL=当前在名单中） + is_active INTEGER DEFAULT 1, -- 1=当前在名单 + source TEXT DEFAULT 'eastmoney_mk0146', + updated_at TEXT + ) + +调度建议： + 每周一次（港股通名单调整频率低，通常每季度例行调整）。 + 例：cron 30 8 * * 1 → hk_connect_list.py + 说明：本脚本是轻量采集（620 只 7 页），不带重型操作。 + +用法： + python3 hk_connect_list.py # 全量翻页采集 + 增量入库 + python3 hk_connect_list.py --show # 打印当前名单统计（只读） +""" +import json +import random +import sqlite3 +import sys +import time +import urllib.request +from datetime import date, datetime +from pathlib import Path + +# ── 配置 ──────────────────────────────────────────────── +DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") # 与同目录脚本一致（daily_kline_collector/market_data） +API_HOST = "push2delay.eastmoney.com" # 延时镜像；push2 主 host 对 246 被封 +API_PATH = "/api/qt/clist/get" +UA = "Mozilla/5.0" +FS = "b:MK0146" # 港股通板块 +PAGE_SIZE = 100 # 单页上限 +MAX_PAGES = 20 # 安全上限（正常 7 页） +PAGE_SLEEP = (1.0, 2.0) # 页间 sleep 1-2 秒（遵守访问频率铁律） +MIN_OK = 500 # 失败保护阈值：采集 < 500 只视为异常，跳过调出标记 +RETRIES = 3 # 单页重试次数 + + +def _fetch_page(pn): + """拉取单页名单，返回原始 JSON（失败返回 None）""" + url = ( + f"https://{API_HOST}{API_PATH}" + f"?pn={pn}&pz={PAGE_SIZE}&po=1&np=1&fltt=2&invt=2&fid=f3" + f"&fs={FS}&fields=f12,f14" + ) + for attempt in range(RETRIES): + try: + req = urllib.request.Request(url, headers={"User-Agent": UA}) + # 显式禁用代理（与 capital_flow_collector/market_data 一致，避免本地代理干扰） + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(req, timeout=10) as r: + text = r.read().decode("utf-8", errors="replace") + return json.loads(text) + except Exception as e: + if attempt < RETRIES - 1: + time.sleep(2) + else: + print(f" 页{pn} 请求失败（{e}）", flush=True) + return None + + +def fetch_all(): + """翻页采集全量名单，返回 (stocks: {code:name}, total_reported: int|None)""" + stocks = {} + total = None + pn = 1 + while pn <= MAX_PAGES: + data = _fetch_page(pn) + if data is None: + break # 页请求失败 → 终止翻页（由上层失败保护兜底） + node = data.get("data") or {} + if total is None and node.get("total"): + total = int(node["total"]) + diff = node.get("diff") or [] + for item in diff: + code = str(item.get("f12") or "").strip() + if not code or not code.isdigit(): + continue + code = f"{int(code):05d}" # 5 位港股代码，如 7 → 00007, 700 → 00700 + stocks[code] = str(item.get("f14") or "").strip() + print(f" 第{pn}页: 本页{len(diff)}只，累计{len(stocks)}只", flush=True) + if len(diff) < PAGE_SIZE: + break # 非满页 = 最后一页 + pn += 1 + if pn <= MAX_PAGES: + time.sleep(random.uniform(*PAGE_SLEEP)) + return stocks, total + + +def init_table(conn): + conn.execute(""" + CREATE TABLE IF NOT EXISTS hk_connect_stocks ( + code TEXT PRIMARY KEY, -- 5位港股代码如 00700 + name TEXT, -- 名称 + in_date TEXT, -- 首次进入名单日期 YYYY-MM-DD + out_date TEXT, -- 调出日期（NULL=当前在名单中） + is_active INTEGER DEFAULT 1, -- 1=当前在名单 + source TEXT DEFAULT 'eastmoney_mk0146', + updated_at TEXT + ) + """) + + +def _singleton_guard(tag): + """单例守卫（铁律5.3）：防重复实例并发写，只跑一个采集实例""" + import fcntl + import os + lock_dir = Path("/tmp/mofin_locks") + lock_dir.mkdir(exist_ok=True) + try: + fd = os.open(str(lock_dir / f"{tag}.lock"), os.O_CREAT | os.O_RDWR) + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd + except OSError: + print(f"[{tag}] 已有实例在运行，退出", file=sys.stderr) + sys.exit(0) + + +def main(): + tag = "hk_connect_list" + if "--show" in sys.argv: + show() + return + _fd = _singleton_guard(tag) + + t0 = time.time() + today = date.today().isoformat() + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[{tag}] {now} 港股通名单采集 开始", flush=True) + + stocks, total = fetch_all() + if not stocks: + print(f"[{tag}] 采集失败：0 只，终止（不触碰 DB）", flush=True) + return 1 + print(f"[{tag}] 本次采集 {len(stocks)} 只（东财 total={total}）", flush=True) + + conn = sqlite3.connect(str(DB_PATH), timeout=30) + init_table(conn) + cur = conn.cursor() + + added, re_added, updated, removed = [], [], [], [] + + # ── 1. UPSERT：本次采集到的标的 ─────────────────── + for code in sorted(stocks): + name = stocks[code] + row = cur.execute( + "SELECT name, is_active FROM hk_connect_stocks WHERE code=?", (code,) + ).fetchone() + if row is None: + # 新出现 → 插入，in_date=今天，is_active=1 + cur.execute( + "INSERT INTO hk_connect_stocks (code, name, in_date, out_date, is_active, source, updated_at) " + "VALUES (?,?,?,NULL,1,?,?)", + (code, name, today, "eastmoney_mk0146", now), + ) + added.append(code) + elif row[1] == 1: + # 已在名单 → 更新 name/updated_at + if name != row[0]: + cur.execute( + "UPDATE hk_connect_stocks SET name=?, updated_at=? WHERE code=? AND is_active=1", + (name, now, code), + ) + updated.append(code) + else: + # 曾调出 → 重新调入：out_date=NULL，is_active=1 + cur.execute( + "UPDATE hk_connect_stocks SET name=?, out_date=NULL, is_active=1, updated_at=? " + "WHERE code=?", + (name, now, code), + ) + re_added.append(code) + + # ── 2. 调出检测：本次缺失但 DB 里 is_active=1 的 ── + # 失败保护：采集结果 < MIN_OK 视为数据源异常，不做调出标记，避免误清空名单 + if len(stocks) < MIN_OK: + print( + f"[{tag}] ⚠ 告警：本次采集仅 {len(stocks)} 只（<{MIN_OK}，正常约620），" + f"跳过调出标记，避免数据源抖动误清空名单", + flush=True, + ) + else: + active_rows = cur.execute( + "SELECT code FROM hk_connect_stocks WHERE is_active=1" + ).fetchall() + missing = [r[0] for r in active_rows if r[0] not in stocks] + for code in missing: + cur.execute( + "UPDATE hk_connect_stocks SET out_date=?, is_active=0, updated_at=? " + "WHERE code=? AND is_active=1", + (today, now, code), + ) + removed.append(code) + + conn.commit() + conn.close() + + # ── 3. 变动日志 ────────────────────────────────── + print(f"[{tag}] 变动汇总：调入 {len(added)} 只，重新调入 {len(re_added)} 只，" + f"更新 {len(updated)} 只，调出 {len(removed)} 只", flush=True) + if added: + print(f" 调入: {','.join(added)}", flush=True) + if re_added: + print(f" 重新调入: {','.join(re_added)}", flush=True) + if removed: + print(f" 调出: {','.join(removed)}", flush=True) + print(f"[{tag}] 完成，耗时 {time.time()-t0:.0f}s", flush=True) + + +def show(): + """--show：打印当前名单统计（只读，不采集）""" + if not DB_PATH.exists(): + print(f"[hk_connect_list] DB 不存在：{DB_PATH}") + return + conn = sqlite3.connect(str(DB_PATH), timeout=10) + init_table(conn) + total = conn.execute("SELECT COUNT(*) FROM hk_connect_stocks").fetchone()[0] + active = conn.execute("SELECT COUNT(*) FROM hk_connect_stocks WHERE is_active=1").fetchone()[0] + today = date.today().isoformat() + added_today = conn.execute( + "SELECT COUNT(*) FROM hk_connect_stocks WHERE in_date=?", (today,) + ).fetchone()[0] + removed_today = conn.execute( + "SELECT COUNT(*) FROM hk_connect_stocks WHERE out_date=?", (today,) + ).fetchone()[0] + conn.close() + print(f"[hk_connect_list] 当前名单统计：累计 {total} 只，在名单 {active} 只，" + f"今日调入 {added_today} 只，今日调出 {removed_today} 只") + + +if __name__ == "__main__": + main() diff --git a/deploy/profile-scripts/market_data.py b/deploy/profile-scripts/market_data.py index 98e7f806..506c82b4 100644 --- a/deploy/profile-scripts/market_data.py +++ b/deploy/profile-scripts/market_data.py @@ -11,7 +11,7 @@ import sqlite3 import urllib.request from pathlib import Path -from market_config import kline_symbol +from market_config import kline_symbol, market_for_code DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") UA = "Mozilla/5.0" @@ -55,12 +55,23 @@ def fetch_tx_klines(code, datalen=120): fetch_sina_klines = fetch_tx_klines -def get_stock_pool(): - """待扫描股票池：stock_daily 的 distinct code（与回测 run_mr_backtest 完全同口径） +def get_stock_pool(market='a'): + """待扫描股票池：按市场参数化（阶段2B 港股接入） - 回测股票池 = SELECT DISTINCT sd.code FROM stock_daily（4266只，含300/688， - 不含301新创业板——数据源未收录）。实盘扫描用同一口径，保证信号 - 覆盖的股票都是回测验证过的。 + 返回 (pool, existing)： + - pool: 待扫描股票池 + - existing: 已有策略/持仓代码（供扫描器跳过） + + market： + - 'a'（默认）: A股池 = stock_daily distinct 6位 code（与回测 run_mr_backtest 完全同口径， + 含300/688，不含301新创业板——数据源未收录）。实盘扫描用同一口径，保证信号 + 覆盖的股票都是回测验证过的。 + - 'hk' : 港股池 = hk_connect_stocks 表 is_active=1 的 code（5位） + + 港股持仓/自选（holdings/holding_strategies 里 5 位代码） + - 'all' : A股池 + 港股池 合并 + + 降级：hk_connect_stocks 表可能不存在（阶段2A 并行开发中）→ try/except + 优雅降级，market='hk' 只返回港股持仓/自选，不报错。 """ conn = sqlite3.connect(str(DB_PATH), timeout=5) try: @@ -72,8 +83,26 @@ def get_stock_pool(): # 与回测完全一致：stock_daily 有K线的股票（回测 universe='a' 排除5位港股） all_stocks = [str(r[0]) for r in conn.execute("SELECT DISTINCT code FROM stock_daily").fetchall()] + # 港股通名单（阶段2A hk_connect_list 采集器写入；表可能尚不存在 → 优雅降级） + hk_connect = set() + try: + for r in conn.execute("SELECT code FROM hk_connect_stocks WHERE is_active=1"): + c = str(r[0]) + if len(c) == 5 and c.isdigit(): + hk_connect.add(c) + except sqlite3.OperationalError as e: + if "no such table" not in str(e).lower(): + raise + # 表不存在（阶段2A 并行开发中）→ 仅港股持仓/自选，不报错 finally: conn.close() # 只留 A 股（6位数字），排除港股（5位0开头）—— 与回测 is_hk_code 逻辑一致 a_stocks = [c for c in all_stocks if len(c) == 6 and c.isdigit()] - return a_stocks, existing + # 港股持仓/自选（5位代码，用 market_for_code 判断） + hk_existing = sorted(c for c in existing if market_for_code(c) == 'hk') + hk_pool = sorted(set(hk_connect) | set(hk_existing)) + if market == 'hk': + return hk_pool, set(hk_existing) + if market == 'all': + return sorted(set(a_stocks) | set(hk_pool)), existing + return a_stocks, existing # market='a'（默认）：与阶段2B 前完全一致