346 lines
12 KiB
Python
346 lines
12 KiB
Python
#!/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):
|
||
"""第四关：资金性质（从腾讯实时行情提取外盘/内盘比）"""
|
||
raw = str(code).strip()
|
||
if raw.startswith(("6", "9")):
|
||
prefix = "sh"
|
||
elif raw.startswith(("0", "3")):
|
||
prefix = "sz"
|
||
else:
|
||
return False, 0, "非A股"
|
||
|
||
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)
|
||
text = r.stdout.decode("gbk", errors="ignore")
|
||
parts = text.split("~")
|
||
if len(parts) < 40:
|
||
return False, 0, "数据不足"
|
||
|
||
# 腾讯字段：[7]=外盘(主动买,股),[8]=内盘(主动卖,股)
|
||
try:
|
||
outer = int(float(parts[7])) if parts[7] else 0 # 外盘
|
||
inner = int(float(parts[8])) if parts[8] else 0 # 内盘
|
||
except:
|
||
return False, 0, "解析失败"
|
||
|
||
if outer <= 0 or inner <= 0:
|
||
return False, 0, "无盘口数据"
|
||
|
||
score = 0
|
||
ratio = outer / inner if inner > 0 else 1
|
||
checks = []
|
||
|
||
if ratio > 1.3:
|
||
score += 2
|
||
checks.append(f"外/内={ratio:.2f}")
|
||
elif ratio > 1.0:
|
||
score += 1
|
||
checks.append(f"买稍强{ratio:.2f}")
|
||
else:
|
||
checks.append(f"卖稍强{ratio:.2f}")
|
||
|
||
# 绝对量也说明资金活跃度
|
||
total = outer + inner
|
||
if total > 50000000: # >5000万股
|
||
score += 1
|
||
checks.append(f"活跃{total/10000:.0f}万")
|
||
|
||
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()
|