214 lines
8.1 KiB
Python
214 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
"""capital_flow_collector.py — 个股资金流数据采集器
|
||
|
||
每30分钟拉一次持仓+自选的超大单/大单/中单/小单资金流向。
|
||
输出到 capital_flow_cache.json 供 price_monitor 和报告使用。
|
||
|
||
API: push2his.eastmoney.com 个股资金流日线
|
||
"""
|
||
import json, os, sys, time, urllib.request
|
||
from datetime import datetime
|
||
from urllib.request import urlopen, Request
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from threading import Semaphore
|
||
from mofin_db import get_conn, write_capital_flow_cache
|
||
|
||
DATA_DIR = "/home/hmo/web-dashboard/data"
|
||
CACHE_PATH = f"{DATA_DIR}/capital_flow_cache.json"
|
||
|
||
UA = "Mozilla/5.0"
|
||
# 限速器：最多5个并发，每请求后强制间隔0.3s
|
||
RATE_LIMIT = Semaphore(5)
|
||
MIN_INTERVAL = 0.3
|
||
_last_req = 0
|
||
|
||
def _rate_limited_request(url, referer="https://data.eastmoney.com/"):
|
||
"""带速率限制的HTTP GET，用Semaphore控制并发数"""
|
||
global _last_req
|
||
with RATE_LIMIT:
|
||
elapsed = time.time() - _last_req
|
||
if elapsed < MIN_INTERVAL:
|
||
time.sleep(MIN_INTERVAL - elapsed)
|
||
proxy_handler = urllib.request.ProxyHandler({})
|
||
opener = urllib.request.build_opener(proxy_handler)
|
||
req = Request(url, headers={"User-Agent": UA, "Referer": referer})
|
||
try:
|
||
resp = opener.open(req, timeout=8)
|
||
_last_req = time.time()
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
except Exception:
|
||
return None
|
||
|
||
# eastmoney secid: 1=上海 0=深圳
|
||
def secid(code):
|
||
code = str(code).strip()
|
||
if code.startswith(("6", "9")):
|
||
return f"1.{code}"
|
||
return f"0.{code}"
|
||
|
||
def fetch_flow(code, days=5):
|
||
"""拉取个股近N日资金流（Sina MoneyFlow ssl_qsfx_zjlrqs，最新在前 —
|
||
eastmoney 在 246 不可达、ssl_qsfx_lscjfb 数据停在2020年（2026-07-22 实证）"""
|
||
code = str(code).strip()
|
||
if code.startswith(("6", "9")):
|
||
dm = f"sh{code}"
|
||
elif len(code) == 5 and code[0] in "01":
|
||
return None # 港股 sina 不支持资金流
|
||
else:
|
||
dm = f"sz{code}"
|
||
url = ("https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/"
|
||
f"MoneyFlow.ssl_qsfx_zjlrqs?daima={dm}")
|
||
data = _rate_limited_request(url, referer="https://finance.sina.com.cn")
|
||
if not data or not isinstance(data, list) or not data:
|
||
return None
|
||
result = []
|
||
for d in data[:days]: # 最新在前，取近N日
|
||
try:
|
||
r0n = float(d.get("r0_net", 0) or 0)
|
||
r1n = float(d.get("r1_net", 0) or 0)
|
||
r2n = float(d.get("r2_net", 0) or 0)
|
||
r3n = float(d.get("r3_net", 0) or 0)
|
||
result.append({
|
||
"date": d.get("opendate", ""),
|
||
"main_net": r0n + r1n,
|
||
"super_large": r0n,
|
||
"large": r1n,
|
||
"medium": r2n,
|
||
"small": r3n,
|
||
})
|
||
except Exception:
|
||
continue
|
||
result.reverse() # 转为时间升序（analyze_flow 以 [-1] 为最新日）
|
||
return result or None
|
||
|
||
def fetch_flow_intraday(code):
|
||
"""拉取当日分时资金流（用于盘中判断）"""
|
||
sid = secid(code)
|
||
url = f"http://push2.eastmoney.com/api/qt/stock/fflow/kline/get?secid={sid}&fields1=f1,f2,f3,f7&fields2=f51,f52,f53,f54,f55,f56,f57&klt=1&lmt=120"
|
||
try:
|
||
resp = urlopen(url, timeout=5)
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
klines = data.get("data", {}).get("klines", [])
|
||
if not klines:
|
||
return None
|
||
latest = klines[-1].split(",")
|
||
return {
|
||
"main_net": float(latest[1]),
|
||
"super_large": float(latest[2]),
|
||
"large": float(latest[3]),
|
||
}
|
||
except:
|
||
return None
|
||
|
||
def analyze_flow(flow_data):
|
||
"""分析资金流模式"""
|
||
if not flow_data or len(flow_data) < 2:
|
||
return {}
|
||
|
||
result = {"alerts": [], "pattern": ""}
|
||
|
||
# 最近两日对比
|
||
d1 = flow_data[-1] # 最新日
|
||
d2 = flow_data[-2] # 前一日
|
||
|
||
# 超大单信号
|
||
sl1 = d1["super_large"]
|
||
sl2 = d2["super_large"]
|
||
|
||
# 连续形态判断
|
||
main_trend = sum(d["main_net"] for d in flow_data[-3:])
|
||
sl_trend = sum(d["super_large"] for d in flow_data[-3:])
|
||
|
||
# 1. 主力连续流入
|
||
if main_trend > 50000000 and sl1 > 0 and sl2 > 0:
|
||
result["pattern"] = "主力持续流入"
|
||
result["alerts"].append("主力连续3日净流入")
|
||
|
||
# 2. 超大单突然转向（连续流入→流出 或 流出→流入）
|
||
if sl1 * sl2 < 0: # 方向反转
|
||
if sl1 > 0 and sl2 < 0:
|
||
result["pattern"] = "超大单由出转入"
|
||
result["alerts"].append("超大单转为净买入(暗示消息即将落地)")
|
||
elif sl1 < 0 and sl2 > 0:
|
||
result["pattern"] = "超大单由入转出"
|
||
result["alerts"].append("超大单转为净卖出(利好出货嫌疑)")
|
||
|
||
# 3. 价格与资金流背离（缺当前价格作比较，在主脚本中完成）
|
||
# 4. 单日暴量
|
||
max_sl = max(abs(d["super_large"]) for d in flow_data)
|
||
if max_sl == abs(sl1) and abs(sl1) > 100000000:
|
||
result["pattern"] = "单日资金暴量"
|
||
result["alerts"].append(f"今日超大单异常: {sl1/100000000:.2f}亿")
|
||
|
||
return result
|
||
|
||
def main():
|
||
codes = set()
|
||
# 读取持仓+自选（从DB直接读，替代已删除的mo_data）
|
||
try:
|
||
import sqlite3
|
||
_db = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
|
||
for row in _db.execute("SELECT DISTINCT code FROM holdings WHERE is_active=1").fetchall():
|
||
if row[0]: codes.add(row[0])
|
||
for row in _db.execute("SELECT DISTINCT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略'").fetchall():
|
||
if row[0]: codes.add(row[0])
|
||
_db.close()
|
||
except:
|
||
pass
|
||
|
||
all_flows = {}
|
||
|
||
# 并行抓取：ThreadPoolExecutor + 内置限速器（Semaphore 5 + 0.3s间隔）
|
||
code_list = sorted(codes)
|
||
if not code_list:
|
||
print("[capital_flow] 无代码需要采集")
|
||
return
|
||
|
||
def fetch_one(code):
|
||
flow = fetch_flow(code, days=5)
|
||
if flow:
|
||
analysis = analyze_flow(flow) or {}
|
||
# 聚合成 build_prompt 需要的字段（此前只有 alerts/pattern，12维资金面恒为0）
|
||
net = round(sum(d["main_net"] for d in flow) / 1e4, 1)
|
||
main_f = round(sum(d["super_large"] for d in flow) / 1e4, 1)
|
||
retail = round(sum(d["medium"] + d["small"] for d in flow) / 1e4, 1)
|
||
analysis["net_flow"] = net
|
||
analysis["main_force"] = main_f
|
||
analysis["retail_flow"] = retail
|
||
analysis["trend"] = analysis.get("pattern") or \
|
||
("主力流入" if main_f > 0 else ("主力流出" if main_f < 0 else "中性"))
|
||
return (code, {
|
||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"flow": flow,
|
||
"analysis": analysis,
|
||
})
|
||
return (code, None)
|
||
|
||
with ThreadPoolExecutor(max_workers=5) as pool:
|
||
futures = {pool.submit(fetch_one, c): c for c in code_list}
|
||
for f in as_completed(futures):
|
||
code, result = f.result()
|
||
if result:
|
||
all_flows[code] = result
|
||
|
||
# 写缓存
|
||
cache = {
|
||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"stocks": all_flows,
|
||
}
|
||
# 写 DB（替代 capital_flow_cache.json）
|
||
conn = get_conn()
|
||
write_capital_flow_cache(conn, cache)
|
||
conn.close()
|
||
print(f"[capital_flow] {len(all_flows)}/{len(code_list)}只更新完成")
|
||
|
||
# 2026-08-13 资金流突变处理（老莫设计）：正面突变→选股分析，负面突变(持仓)→重评+XMPP报告
|
||
try:
|
||
from fund_flow_alert import process_flow_alerts
|
||
process_flow_alerts(all_flows)
|
||
except Exception as e:
|
||
print(f"[资金流突变处理异常] {e}", flush=True)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|