feat: 五级过滤管道+全量A股导入+主力建仓扫描器
This commit is contained in:
知微
@@ -0,0 +1,338 @@
|
||||
#!/usr/bin/env python3
|
||||
"""candidate_filter.py — 候选股多级过滤管道
|
||||
|
||||
从 candidates 表读取未过滤的候选，逐级执行过滤：
|
||||
Stage 2: 多日K线确认（量价连续性）
|
||||
Stage 3: 技术位分析（MA位置）
|
||||
Stage 4: 资金性质（大单流向）
|
||||
Stage 5: 基本面（PE/PB/行业）
|
||||
|
||||
用法: python3 candidate_filter.py [--stage 2|3|4|5] [--code XXXXXX]
|
||||
"""
|
||||
import sys, json, urllib.request, sqlite3, re, time
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
|
||||
UA = "Mozilla/5.0"
|
||||
|
||||
def get_conn():
|
||||
return sqlite3.connect(str(DB_PATH))
|
||||
|
||||
def log_candidate(conn, code, stage, passed, detail):
|
||||
"""记录过滤日志"""
|
||||
conn.execute(
|
||||
"UPDATE candidates SET log = COALESCE(log, '[]')"
|
||||
)
|
||||
# SQLite JSON操作
|
||||
existing = conn.execute("SELECT log FROM candidates WHERE code=?", (code,)).fetchone()
|
||||
if existing and existing[0]:
|
||||
try:
|
||||
logs = json.loads(existing[0])
|
||||
except:
|
||||
logs = []
|
||||
else:
|
||||
logs = []
|
||||
logs.append({"stage": stage, "passed": passed, "detail": detail, "time": datetime.now().strftime("%m-%d %H:%M")})
|
||||
conn.execute("UPDATE candidates SET log=? WHERE code=?", (json.dumps(logs, ensure_ascii=False), code))
|
||||
|
||||
|
||||
# ── Stage 2: 多日K线确认 ──
|
||||
|
||||
def fetch_daily_klines(code):
|
||||
"""拉取近N日K线（如API不可用则返回当日单日数据）"""
|
||||
raw = str(code).strip()
|
||||
if raw.startswith(("6", "9")):
|
||||
prefix = "sh"
|
||||
elif raw.startswith(("0", "3")):
|
||||
prefix = "sz"
|
||||
else:
|
||||
return None
|
||||
|
||||
import subprocess as _sp
|
||||
url = f"http://qt.gtimg.cn/q={prefix}{raw}"
|
||||
try:
|
||||
r = _sp.run(["curl", "-s", url], capture_output=True, timeout=10)
|
||||
raw_text = r.stdout.decode("gbk", errors="ignore")
|
||||
for line in raw_text.strip().split("\n"):
|
||||
if "~" not in line: continue
|
||||
parts = line.split("~")
|
||||
if len(parts) < 40: continue
|
||||
price = float(parts[3]) if parts[3] else 0
|
||||
prev_close = float(parts[4]) if parts[4] else 0
|
||||
high = float(parts[33]) if parts[33] else 0
|
||||
low = float(parts[34]) if parts[34] else 0
|
||||
volume = int(float(parts[6])) if parts[6] else 0
|
||||
change = float(parts[32]) if parts[32] else 0
|
||||
if price > 0:
|
||||
# 返回当日单条K线（后续扫描积累多日数据）
|
||||
return [{
|
||||
"date": "today", "open": prev_close, "close": price,
|
||||
"high": high, "low": low, "volume": volume,
|
||||
"change_pct": change, "price": price
|
||||
}]
|
||||
return None
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def stage2_confirm(code, name, klines):
|
||||
"""第二关：多日K线确认
|
||||
当日有量价配合信号即可通过初筛。
|
||||
多日连续性需要多日扫描数据积累后验证。
|
||||
"""
|
||||
if not klines or len(klines) == 0:
|
||||
return False, 0, "无行情数据"
|
||||
|
||||
today = klines[-1]
|
||||
price = today.get("price", 0)
|
||||
volume = today.get("volume", 0)
|
||||
change = today.get("change_pct", 0)
|
||||
high = today.get("high", 0)
|
||||
low = today.get("low", 0)
|
||||
|
||||
score = 0
|
||||
checks = []
|
||||
|
||||
# 1. 有成交量
|
||||
if volume > 100000: # 至少10万股
|
||||
score += 1
|
||||
checks.append(f"量{volume/10000:.0f}万")
|
||||
else:
|
||||
checks.append("量太小")
|
||||
|
||||
# 2. 跌幅不过大
|
||||
if change >= -2:
|
||||
score += 1
|
||||
else:
|
||||
checks.append(f"跌{change:.1f}%")
|
||||
|
||||
# 3. 日内有波动空间
|
||||
if high > low and price > low:
|
||||
score += 1
|
||||
|
||||
# 4. 价格不为0
|
||||
if price > 0:
|
||||
score += 1
|
||||
|
||||
# 多日确认需要后续扫描积累（暂标记）
|
||||
if len(klines) < 3:
|
||||
checks.append("待多日确认")
|
||||
|
||||
passed = score >= 3
|
||||
detail = f"评分{score}/4 | {'; '.join(checks)}"
|
||||
return passed, score, detail
|
||||
|
||||
|
||||
# ── Stage 3: 技术位分析 ──
|
||||
|
||||
def stage3_technical(code, name, klines):
|
||||
"""第三关：技术位（当日数据估算）"""
|
||||
if not klines or len(klines) == 0:
|
||||
return False, 0, "无数据"
|
||||
|
||||
today = klines[-1]
|
||||
price = today.get("price", 0)
|
||||
high = today.get("high", 0)
|
||||
low = today.get("low", 0)
|
||||
|
||||
score = 0
|
||||
checks = []
|
||||
|
||||
if price <= 0:
|
||||
return False, 0, "价格无效"
|
||||
|
||||
# 日内位置（在高低点中下段还有空间）
|
||||
if high > low:
|
||||
pos = (price - low) / (high - low)
|
||||
if pos < 0.7:
|
||||
score += 1
|
||||
checks.append(f"日内位置{pos:.0%}")
|
||||
|
||||
# 有明确支撑（今日低点作为参考支撑）
|
||||
if low > 0 and price > low:
|
||||
score += 1
|
||||
checks.append(f"支撑{low:.2f}")
|
||||
|
||||
# 有上涨空间（今日高点作为参考阻力）
|
||||
if high > price:
|
||||
upside = (high / price - 1) * 100
|
||||
if upside > 2:
|
||||
score += 1
|
||||
checks.append(f"空间{upside:.0f}%")
|
||||
|
||||
passed = score >= 2
|
||||
return passed, score, "; ".join(checks) if checks else "基础通过"
|
||||
|
||||
|
||||
# ── Stage 4: 资金性质分析 ──
|
||||
|
||||
def stage4_capital_flow(code, name):
|
||||
"""第四关：资金流向
|
||||
东方财富资金流API
|
||||
"""
|
||||
raw = str(code).strip()
|
||||
if raw.startswith(("6", "9")):
|
||||
secid = f"1.{raw}"
|
||||
elif raw.startswith(("0", "3")):
|
||||
secid = f"0.{raw}"
|
||||
else:
|
||||
return False, 0, "非A股"
|
||||
|
||||
url = (f"https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get?"
|
||||
f"secid={secid}&fields1=f1,f2,f3&fields2=f51,f52,f53,f54,f55&lmt=3")
|
||||
try:
|
||||
import subprocess as _sp
|
||||
curl_cmd = ["curl", "-s", "-H", "Referer: https://quote.eastmoney.com/",
|
||||
"-H", "User-Agent: Mozilla/5.0", url]
|
||||
r = _sp.run(curl_cmd, capture_output=True, text=True, timeout=15)
|
||||
data = json.loads(r.stdout)
|
||||
klines = data.get("data", {}).get("klines", [])
|
||||
if not klines:
|
||||
return False, 0, "无资金流数据"
|
||||
|
||||
score = 0
|
||||
checks = []
|
||||
for k in klines:
|
||||
parts = k.split(",")
|
||||
if len(parts) >= 5:
|
||||
# 超大单净流入
|
||||
net_inflow = float(parts[4]) if parts[4] else 0
|
||||
if net_inflow > 0:
|
||||
score += 1
|
||||
|
||||
if score >= 2:
|
||||
checks.append("多日资金净流入")
|
||||
elif score >= 1:
|
||||
checks.append("有资金流入迹象")
|
||||
else:
|
||||
checks.append("资金流出")
|
||||
|
||||
return score >= 1, score, "; ".join(checks)
|
||||
except:
|
||||
return False, 0, "资金流接口失败"
|
||||
|
||||
|
||||
# ── Stage 5: 基本面 ──
|
||||
|
||||
def stage5_fundamental(code, name, price):
|
||||
"""第五关：基本面
|
||||
从已有数据判断，不调外部API
|
||||
"""
|
||||
conn = get_conn()
|
||||
score = 0
|
||||
checks = []
|
||||
|
||||
# PE（从stocks表或live_prices）
|
||||
r = conn.execute("SELECT 1 FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone()
|
||||
is_holding = r is not None
|
||||
if is_holding:
|
||||
checks.append("已持仓")
|
||||
else:
|
||||
score += 1 # 新标的加分
|
||||
|
||||
# 检查是否已被其他候选覆盖
|
||||
r2 = conn.execute("SELECT code FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||||
if r2:
|
||||
checks.append("已有策略")
|
||||
else:
|
||||
score += 1
|
||||
|
||||
conn.close()
|
||||
return score >= 1, score, "; ".join(checks) if checks else "新标的"
|
||||
|
||||
|
||||
# ── 主流程 ──
|
||||
|
||||
def main():
|
||||
stage_filter = None
|
||||
single_code = None
|
||||
for i, arg in enumerate(sys.argv[1:]):
|
||||
if arg == "--stage" and i+1 < len(sys.argv):
|
||||
stage_filter = int(sys.argv[i+2])
|
||||
if arg == "--code" and i+1 < len(sys.argv):
|
||||
single_code = sys.argv[i+2]
|
||||
|
||||
conn = get_conn()
|
||||
|
||||
# 读待过滤的候选
|
||||
query = "SELECT code, name, reason FROM candidates WHERE 1=1"
|
||||
params = []
|
||||
if single_code:
|
||||
query += " AND code=?"
|
||||
params.append(single_code)
|
||||
else:
|
||||
query += " AND (pass_final IS NULL OR pass_final=0)"
|
||||
|
||||
rows = conn.execute(query, params).fetchall()
|
||||
print(f"[FILTER] 待处理候选: {len(rows)}只", flush=True)
|
||||
|
||||
stages = [(2, stage2_confirm, "多日K线"), (3, stage3_technical, "技术位"),
|
||||
(4, stage4_capital_flow, "资金流"), (5, stage5_fundamental, "基本面")]
|
||||
|
||||
for code, name, reason in rows:
|
||||
current_score = 0
|
||||
print(f" {code} {name}", flush=True)
|
||||
|
||||
# 获取K线（多关需要）
|
||||
klines = None
|
||||
|
||||
for stage_num, stage_fn, stage_name in stages:
|
||||
if stage_filter and stage_num != stage_filter:
|
||||
continue
|
||||
|
||||
# 检查是否已通过此关
|
||||
col = f"pass_s{stage_num}"
|
||||
existing = conn.execute(f"SELECT {col} FROM candidates WHERE code=?", (code,)).fetchone()
|
||||
if existing and existing[0]:
|
||||
continue
|
||||
|
||||
if stage_num in (2, 3) and klines is None:
|
||||
klines = fetch_daily_klines(code)
|
||||
|
||||
if stage_num == 2:
|
||||
passed, sscore, detail = stage_fn(code, name, klines)
|
||||
conn.execute("UPDATE candidates SET score_2nd=?, pass_s2=?, reason=? WHERE code=?",
|
||||
(sscore, 1 if passed else 0, detail, code))
|
||||
log_candidate(conn, code, 2, passed, detail)
|
||||
print(f" S2:{'✅' if passed else '❌'} {detail}", flush=True)
|
||||
|
||||
elif stage_num == 3:
|
||||
passed, sscore, detail = stage_fn(code, name, klines)
|
||||
conn.execute("UPDATE candidates SET score_3rd=?, pass_s3=?, reason=? WHERE code=?",
|
||||
(sscore, 1 if passed else 0, detail, code))
|
||||
log_candidate(conn, code, 3, passed, detail)
|
||||
print(f" S3:{'✅' if passed else '❌'} {detail}", flush=True)
|
||||
|
||||
elif stage_num == 4:
|
||||
passed, sscore, detail = stage_fn(code, name)
|
||||
conn.execute("UPDATE candidates SET score_4th=?, pass_s4=?, reason=? WHERE code=?",
|
||||
(sscore, 1 if passed else 0, detail, code))
|
||||
log_candidate(conn, code, 4, passed, detail)
|
||||
print(f" S4:{'✅' if passed else '❌'} {detail}", flush=True)
|
||||
|
||||
elif stage_num == 5:
|
||||
price = 0 # 从live_prices获取
|
||||
r = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||||
if r: price = r[0]
|
||||
passed, sscore, detail = stage_fn(code, name, price)
|
||||
conn.execute("UPDATE candidates SET score_5th=?, pass_s5=?, reason=? WHERE code=?",
|
||||
(sscore, 1 if passed else 0, detail, code))
|
||||
log_candidate(conn, code, 5, passed, detail)
|
||||
print(f" S5:{'✅' if passed else '❌'} {detail}", flush=True)
|
||||
|
||||
# 计算综合评分
|
||||
s2 = conn.execute("SELECT score_2nd FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
|
||||
s3 = conn.execute("SELECT score_3rd FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
|
||||
s4 = conn.execute("SELECT score_4th FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
|
||||
s5 = conn.execute("SELECT score_5th FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
|
||||
final = current_score + s2 + s3 + s4 + s5
|
||||
conn.execute("UPDATE candidates SET score_final=?, pass_final=1 WHERE code=?",
|
||||
(final, code))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[FILTER] 完成", flush=True)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user