diff --git a/deploy/profile-scripts/candidate_filter.py b/deploy/profile-scripts/candidate_filter.py index be443cf4..cb927d03 100644 --- a/deploy/profile-scripts/candidate_filter.py +++ b/deploy/profile-scripts/candidate_filter.py @@ -48,33 +48,28 @@ def log_candidate(conn, code, stage, passed, detail): # ── Stage 2: 多日K线确认 ── -def fetch_daily_klines(code): - """拉取近10日日K线(Sina 240分钟线=日K)""" +def fetch_daily_klines(code, conn=None): + """读 stock_daily 近10日日K(2026-08-24 数据分层铁律:不再curl sina,DB千万行日线本地读,毫秒级)""" 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" + own = conn is None + if own: + conn = get_conn() try: - r = _sp.run(["curl", "-s", "--noproxy", "*", url], capture_output=True, timeout=10) - data = _json.loads(r.stdout) - if not data: + rows = conn.execute( + "SELECT date, open, close, high, low, volume FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 10", + (raw,)).fetchall() + if not rows: return None result = [] - for k in data: + for r in reversed(rows): # 日期升序(与原sina返回一致) 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"]), + "date": r[0], + "open": float(r[1] or 0), + "close": float(r[2] or 0), + "high": float(r[3] or 0), + "low": float(r[4] or 0), + "volume": int(r[5] or 0), + "price": float(r[2] or 0), "change_pct": 0, }) # 计算涨跌幅 @@ -83,9 +78,11 @@ def fetch_daily_klines(code): if prev > 0: result[i]["change_pct"] = (result[i]["close"] / prev - 1) * 100 return result - except Exception as e: - return None + except Exception: return None + finally: + if own: + conn.close() def stage2_confirm(code, name, klines): @@ -184,56 +181,41 @@ def stage3_technical(code, name, klines): # ── Stage 4: 资金性质分析 ── def stage4_capital_flow(code, name): - """第四关:资金性质(从腾讯实时行情提取外盘/内盘比)""" + """第四关:资金性质(读 capital_flow_cache 主力净流入,2026-08-24 数据分层铁律: + 不再curl腾讯quote。主力净流入语义强于外盘/内盘比——直接回答"主力在买还是卖")""" 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}" + conn = get_conn() 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, "无盘口数据" - + r = conn.execute("SELECT cache_json FROM capital_flow_cache ORDER BY id DESC LIMIT 1").fetchone() + if not r: + return False, 0, "无资金流缓存" + import json as _j + stocks = (_j.loads(r[0]) or {}).get("stocks") or {} + info = stocks.get(raw) + if not info or not info.get("flow"): + return False, 0, "无资金流数据" + flow = info["flow"] # [{date, main_net, super_large, ...}] 日期升序 + recent = flow[-3:] # 近3日 score = 0 - ratio = outer / inner if inner > 0 else 1 checks = [] - - if ratio > 1.3: + pos_days = sum(1 for f in recent if (f.get("main_net") or 0) > 0) + total_net = sum((f.get("main_net") or 0) for f in recent) + if pos_days >= 2: score += 2 - checks.append(f"外/内={ratio:.2f}") - elif ratio > 1.0: + checks.append(f"主力净流入{pos_days}/3日") + elif pos_days >= 1: score += 1 - checks.append(f"买稍强{ratio:.2f}") + checks.append(f"净流入{pos_days}/3日") else: - checks.append(f"卖稍强{ratio:.2f}") - - # 绝对量也说明资金活跃度 - total = outer + inner - if total > 50000000: # >5000万股 + checks.append("连续净流出") + if total_net > 30000000: # 3日累计>3000万 score += 1 - checks.append(f"活跃{total/10000:.0f}万") - + checks.append(f"累计{total_net/10000:.0f}万") return score >= 1, score, "; ".join(checks) - except: - return False, 0, "接口失败" + except Exception as e: + return False, 0, f"读取失败:{str(e)[:30]}" + finally: + conn.close() # ── Stage 5: 基本面 ──