feat: 港股接入阶段2——hk_connect_list港股通名单采集器(东财MK0146板块620只,push2delay镜像避开主host封禁)+get_stock_pool(market=)参数化(默认A股行为不变)+daily_kline_collector港股批
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
@@ -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 前完全一致
|
||||
|
||||
Reference in New Issue
Block a user