feat: multi-strategy parallel screening (balanced_alpha+dual_low+quality_value)

mo_alphasift_bridge.py

@@ -1,35 +1,27 @@
 #!/usr/bin/env python3
 """
-mo_alphasift_bridge.py — AlphaSift 选股 → MoFin 自选池自动对接
+mo_alphasift_bridge.py — AlphaSift 多策略并行选股 → MoFin 自选池
 
-流程：
+支持同时跑多个策略，合并去重后写入自选池。
-1. 提交 AlphaSift 异步选股任务 → 轮询结果
+默认三策略: balanced_alpha + dual_low + quality_value
-2. 过滤（评分阈值 + 去重）
-3. 写入 MoFin watchlist.json（含来源/日期/策略备注）
-4. 调用 regenerate_all() 生成策略
-5. price_monitor 自动接管
 
-用法：
+用法:
-  python3 mo_alphasift_bridge.py                     # 默认策略
+  python3 mo_alphasift_bridge.py                              # 默认三策略
-  python3 mo_alphasift_bridge.py --dry-run            # 只看不写
+  python3 mo_alphasift_bridge.py --strategy dual_low          # 单策略
-  python3 mo_alphasift_bridge.py --min-score 6        # 最低评分
+  python3 mo_alphasift_bridge.py --dry-run                     # 只看不写
-  python3 mo_alphasift_bridge.py --strategy list      # 列出所有策略
+  python3 mo_alphasift_bridge.py --strategy list               # 列出所有策略
-
-cron: 0 10 * * 1-5 cd /home/hmo/MoFin && python3 mo_alphasift_bridge.py
 """
 
 import sys, os, json, argparse, urllib.request, time
 from datetime import datetime
 from pathlib import Path
 
-# ── 配置 ─────────────────────────────────────────────────────────────
-
 DSA_API = "http://127.0.0.1:8001"
 MOFIN_DATA = Path("/home/hmo/web-dashboard/data")
 WATCHLIST_PATH = MOFIN_DATA / "watchlist.json"
 PORTFOLIO_PATH = MOFIN_DATA / "portfolio.json"
 
-DEFAULT_STRATEGY = "balanced_alpha"
+DEFAULT_STRATEGIES = "balanced_alpha,dual_low,quality_value"
 DEFAULT_MARKET = "cn"
 DEFAULT_MAX = 15
 MIN_SCORE = 5
@@ -39,17 +31,13 @@ POLL_TIMEOUT = 300
 
 
 def load_json(path):
-    try:
+    try: return json.loads(path.read_text(encoding="utf-8"))
-        return json.loads(path.read_text(encoding="utf-8"))
+    except: return None
-    except:
-        return None
-
 
 def save_json(path, data):
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
 
-
 def api(endpoint, method="GET", body=None):
     url = f"{DSA_API}{endpoint}"
     data = json.dumps(body).encode() if body else None
@@ -59,99 +47,119 @@ def api(endpoint, method="GET", body=None):
         with urllib.request.urlopen(req, timeout=30) as r:
             return json.loads(r.read())
     except Exception as e:
-        print(f"  API 错误: {e}")
+        print(f"  API错误: {e}")
         return None
 
-
 def get_existing_codes():
     codes = set()
     for path in [WATCHLIST_PATH, PORTFOLIO_PATH]:
         data = load_json(path)
-        if not data:
+        if not data: continue
-            continue
         key = "stocks" if "watchlist" in str(path) else "holdings"
         for item in data.get(key, []):
             c = str(item.get("code", "")).strip()
-            if c:
+            if c: codes.add(c)
-                codes.add(c)
     return codes
 
 
-def run_screen(strategy, market, max_results, dry_run=False):
+def run_one_strategy(strategy, market, max_results):
-    now = datetime.now()
+    """跑单个策略，返回候选股列表"""
-    date_str = now.strftime("%Y-%m-%d")
+    print(f"\n{'='*50}")
-    time_str = now.strftime("%Y-%m-%d %H:%M")
+    print(f"策略: {strategy}")
+    print(f"{'='*50}")
     
-    # 1. 提交异步任务
-    print(f"AlphaSift 选股: {strategy} (异步)", flush=True)
     task = api("/api/v1/alphasift/screen/tasks", "POST", {
         "strategy": strategy, "market": market, "max_results": max_results
     })
     if not task or not task.get("task_id"):
-        print("  FAIL: 提交任务失败")
+        print(f"  FAIL: 提交失败")
-        return
+        return []
     
     task_id = task["task_id"]
-    print(f"  任务ID: {task_id[:12]}...", flush=True)
+    print(f"  任务: {task_id[:12]}...", flush=True)
     
-    # 2. 轮询
     waited = 0
-    result = None
     while waited < POLL_TIMEOUT:
         time.sleep(POLL_INTERVAL)
         waited += POLL_INTERVAL
         status = api(f"/api/v1/alphasift/screen/tasks/{task_id}")
-        if not status:
+        if not status: continue
-            continue
         s = status.get("status", "")
         if s == "completed":
             print(f"  完成 ({waited}s)")
             result = status.get("result", {})
-            break
+            candidates = result.get("candidates", [])
+            print(f"  候选: {len(candidates)} 只")
+            if candidates:
+                for c in candidates[:3]:
+                    print(f"    {c.get('code','?')} {c.get('name',c.get('title','?'))} 评分{c.get('score','?'):.1f}")
+            return candidates
         elif s == "failed":
-            print(f"  FAIL: {status.get('error', '')}")
+            print(f"  FAIL: {status.get('error','')}")
-            return
+            return []
         else:
-            pct = status.get("progress", 0)
+            if waited % 60 == 0:
-            print(f"  ...{s} ({pct}%)", flush=True)
+                print(f"  ...{s} ({status.get('progress',0)}%)", flush=True)
-    else:
+    
-        print(f"  FAIL: 超时 ({POLL_TIMEOUT}s)")
+    print(f"  FAIL: 超时")
+    return []
+
+
+def run_all(strategies_str, market, max_results, dry_run=False):
+    """多策略并行 → 合并去重 → MoFin 自选池"""
+    strategies = [s.strip() for s in strategies_str.split(",") if s.strip()]
+    now = datetime.now()
+    date_str = now.strftime("%Y-%m-%d")
+    time_str = now.strftime("%Y-%m-%d %H:%M")
+    
+    print(f"AlphaSift 多策略选股: {', '.join(strategies)}")
+    print(f"开始: {time_str}")
+    
+    # 逐个跑策略，汇总
+    all_candidates = []
+    seen = set()
+    for strategy in strategies:
+        candidates = run_one_strategy(strategy, market, max_results)
+        for c in candidates:
+            code = str(c.get("code", "")).strip()
+            if code in seen: continue
+            seen.add(code)
+            c["_strategy"] = strategy
+            all_candidates.append(c)
+    
+    if not all_candidates:
+        print("\n无候选股")
         return
     
-    candidates = result.get("candidates", [])
+    print(f"\n汇总: {len(all_candidates)} 只候选股 (去重后)")
-    if not candidates:
+    for s in strategies:
-        print("  无候选股")
+        cnt = sum(1 for c in all_candidates if c.get("_strategy") == s)
-        return
+        print(f"  {s}: {cnt} 只")
     
-    market_view = result.get("llm_market_view", "")
+    # 过滤
-    print(f"  返回 {len(candidates)} 只候选股")
-    if market_view:
-        print(f"  市场观点: {market_view[:100]}...")
-    
-    # 3. 过滤 + 去重
     existing = get_existing_codes()
     new_stocks = []
-    skipped = 0
+    skipped_score = 0
+    skipped_dup = 0
     
-    for c in candidates:
+    for c in all_candidates:
         code = str(c.get("code", "")).strip()
         score = c.get("score", 0) or c.get("llm_score", 0) or 0
+        if score < MIN_SCORE:
+            skipped_score += 1; continue
+        if code in existing:
+            skipped_dup += 1; continue
+        
         name = c.get("name", "") or c.get("title", "") or code
         reason = c.get("reason", "") or c.get("llm_thesis", "")
-        catalysts = c.get("llm_catalysts", "")
+        src = c.get("_strategy", "unknown")
-        risks = c.get("llm_risks", "")
         
-        if score < MIN_SCORE:
+        factors = c.get("factor_scores", {})
-            skipped += 1
+        factor_note = ", ".join(f"{k}={v:.0f}" for k,v in list(factors.items())[:3]) if factors else ""
-            continue
-        if code in existing:
-            continue
         
-        notes_parts = [f"AlphaSift/{strategy} 评分{score}"]
+        notes = f"AlphaSift/{src} 评分{score:.0f}"
-        if reason:
+        if factor_note: notes += f" [{factor_note}]"
-            notes_parts.append(reason[:150])
+        if reason: notes += f" | {reason[:120]}"
-        if catalysts:
-            notes_parts.append(f"催化剂:{catalysts[:100]}")
         
         new_stocks.append({
             "code": code,
@@ -159,83 +167,74 @@ def run_screen(strategy, market, max_results, dry_run=False):
             "price": c.get("price", 0),
             "source": "alpha_sift",
             "source_detail": {
-                "strategy": strategy,
+                "strategy": src,
+                "strategies_run": strategies,
                 "score": score,
+                "factor_scores": factors,
                 "date": date_str,
                 "reason": reason[:300],
-                "catalysts": catalysts[:200] if catalysts else "",
-                "risks": risks[:200] if risks else "",
-                "market_view": market_view[:300] if market_view else "",
             },
-            "notes": " | ".join(notes_parts),
+            "notes": notes,
             "added_at": time_str,
             "added_by": "AlphaSift",
             "analysis": {},
         })
         existing.add(code)
     
-    print(f"  过滤: {len(new_stocks)} 新标的 (跳过{skipped}只评分不足)")
+    print(f"\n过滤: {len(new_stocks)} 新标的 (评分不足{skipped_score} + 重复{skipped_dup})")
     
     if not new_stocks:
-        print("  无符合条件的新标的")
+        print("无符合条件的新标的")
         return
     
     new_stocks = new_stocks[:MAX_ADD]
     
-    print(f"\n  新增 {len(new_stocks)} 只到自选池:")
+    print(f"\n新增 {len(new_stocks)} 只到自选池:")
     for s in new_stocks:
-        print(f"    {s['code']} {s['name']} (评分{s['source_detail']['score']})")
+        sd = s["source_detail"]
-        if s['source_detail'].get('reason'):
+        print(f"  {s['code']} {s['name']} ({sd['strategy']} 评分{sd['score']:.0f})")
-            print(f"      {s['source_detail']['reason'][:120]}")
     
     if dry_run:
-        print("\n  [DRY RUN] 未写入")
+        print("\n[DRY RUN] 未写入")
         return
     
-    # 4. 写入
+    # 写入
     wl = load_json(WATCHLIST_PATH) or {"stocks": []}
     wl["stocks"].extend(new_stocks)
     wl["updated_at"] = time_str
     save_json(WATCHLIST_PATH, wl)
-    print(f"\n  已写入 {WATCHLIST_PATH}")
+    print(f"\n已写入 {WATCHLIST_PATH}")
     
-    # 5. 策略生成
+    # 策略生成
-    print("\n  调用 regenerate_all() 生成策略...")
+    print("\n调用 regenerate_all()...")
     try:
         sys.path.insert(0, str(MOFIN_DATA.parent))
         from strategy_lifecycle import regenerate_all
         r = regenerate_all(stdout=True)
-        if r:
+        if r: print(f"完成: {r.get('ok',0)}/{r.get('total',0)} 只策略已生成")
-            print(f"  完成: {r.get('ok',0)}/{r.get('total',0)} 只策略已生成")
-            print(f"  price_monitor 将在 2 分钟内自动开始监控")
     except Exception as e:
-        print(f"  WARN: regenerate_all 失败: {e}")
+        print(f"WARN: {e}")
-        print(f"  股票已加入自选池，下次 cron 会自动处理")
 
 
 def list_strategies():
     r = api("/api/v1/alphasift/strategies")
     if r and r.get("strategies"):
-        print("AlphaSift 策略:")
         for s in r["strategies"]:
-            print(f"  {s['id']:20s} {s.get('name','?'):15s} {s.get('description','')[:60]}")
+            print(f"  {s['id']:22s} {s.get('name','?'):10s} {s.get('description','')[:60]}")
 
 
 def main():
     global MIN_SCORE
-    parser = argparse.ArgumentParser(description="AlphaSift → MoFin")
+    p = argparse.ArgumentParser(description="AlphaSift → MoFin")
-    parser.add_argument("--strategy", default=DEFAULT_STRATEGY)
+    p.add_argument("--strategy", default=DEFAULT_STRATEGIES)
-    parser.add_argument("--market", default=DEFAULT_MARKET)
+    p.add_argument("--market", default=DEFAULT_MARKET)
-    parser.add_argument("--max", type=int, default=DEFAULT_MAX)
+    p.add_argument("--max", type=int, default=DEFAULT_MAX)
-    parser.add_argument("--min-score", type=int, default=MIN_SCORE)
+    p.add_argument("--min-score", type=int, default=MIN_SCORE)
-    parser.add_argument("--dry-run", action="store_true")
+    p.add_argument("--dry-run", action="store_true")
-    args = parser.parse_args()
+    args = p.parse_args()
     MIN_SCORE = args.min_score
-    
+    if args.strategy == "list": list_strategies()
-    if args.strategy == "list":
+    else: run_all(args.strategy, args.market, args.max, args.dry_run)
-        list_strategies()
-        return
-    run_screen(args.strategy, args.market, args.max, args.dry_run)
 
 
 if __name__ == "__main__":