diff --git a/scripts/candidate_filter.py b/scripts/candidate_filter.py
new file mode 100644
index 00000000..76034b7a
--- /dev/null
+++ b/scripts/candidate_filter.py
@@ -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()