355 lines
12 KiB
Python
355 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():
|
||
c = sqlite3.connect(str(DB_PATH), timeout=30)
|
||
c.execute("PRAGMA busy_timeout=30000")
|
||
return c
|
||
|
||
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):
|
||
"""拉取近10日日K线（Sina 240分钟线=日K）"""
|
||
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, json as _json
|
||
url = f"http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={prefix}{raw}&scale=240&ma=5&datalen=10"
|
||
try:
|
||
r = _sp.run(["curl", "-s", "--noproxy", "*", url], capture_output=True, timeout=10)
|
||
data = _json.loads(r.stdout)
|
||
if not data:
|
||
return None
|
||
result = []
|
||
for k in data:
|
||
result.append({
|
||
"date": k.get("day", "")[:10],
|
||
"open": float(k["open"]),
|
||
"close": float(k["close"]),
|
||
"high": float(k["high"]),
|
||
"low": float(k["low"]),
|
||
"volume": int(k["volume"]),
|
||
"price": float(k["close"]),
|
||
"change_pct": 0,
|
||
})
|
||
# 计算涨跌幅
|
||
for i in range(1, len(result)):
|
||
prev = result[i-1]["close"]
|
||
if prev > 0:
|
||
result[i]["change_pct"] = (result[i]["close"] / prev - 1) * 100
|
||
return result
|
||
except Exception as e:
|
||
return None
|
||
return None
|
||
|
||
|
||
def stage2_confirm(code, name, klines):
|
||
"""第二关：多日K线确认
|
||
检查：多日量价配合、建仓特征
|
||
"""
|
||
if not klines or len(klines) < 3:
|
||
return False, 0, "K线不足3日"
|
||
|
||
recent = klines[-5:] # 最近5日
|
||
score = 0
|
||
checks = []
|
||
|
||
# 1. 成交量连续递增
|
||
vols = [k["volume"] for k in recent]
|
||
vol_rising = sum(1 for i in range(len(vols)-1) if vols[i] < vols[i+1])
|
||
if vol_rising >= 3:
|
||
score += 2
|
||
checks.append(f"量增{vol_rising}/4日")
|
||
elif vol_rising >= 2:
|
||
score += 1
|
||
checks.append(f"量微增{vol_rising}/4日")
|
||
|
||
# 2. 涨放量、跌缩量
|
||
up_vol = sum(k["volume"] for k in recent if k["change_pct"] >= 0)
|
||
down_vol = sum(k["volume"] for k in recent if k["change_pct"] < 0)
|
||
if down_vol > 0 and up_vol / down_vol > 1.5:
|
||
score += 2
|
||
checks.append(f"涨量/跌量={up_vol/down_vol:.1f}")
|
||
elif down_vol > 0 and up_vol / down_vol > 1:
|
||
score += 1
|
||
|
||
# 3. 价格趋势
|
||
closes = [k["close"] for k in recent]
|
||
up_days = sum(1 for i in range(1, len(closes)) if closes[i] > closes[i-1])
|
||
if up_days >= 3:
|
||
score += 2
|
||
checks.append(f"涨{up_days}/4日")
|
||
elif up_days >= 2:
|
||
score += 1
|
||
|
||
# 4. 无异常放量（单日>3倍均量=可能出货）
|
||
avg_vol = sum(vols) / len(vols) if vols else 1
|
||
max_ratio = max(v / avg_vol for v in vols) if avg_vol > 0 else 1
|
||
if max_ratio < 2.5:
|
||
score += 1
|
||
else:
|
||
checks.append(f"异常量{max_ratio:.0f}倍")
|
||
|
||
passed = score >= 4
|
||
detail = f"评分{score}/7 | {'; '.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()
|