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