#!/usr/bin/env python3 """ mo_alphasift_bridge.py — AlphaSift 选股 → MoFin 自选池自动对接 流程： 1. 提交 AlphaSift 异步选股任务 → 轮询结果 2. 过滤（评分阈值 + 去重） 3. 写入 MoFin watchlist.json（含来源/日期/策略备注） 4. 调用 regenerate_all() 生成策略 5. price_monitor 自动接管 用法： python3 mo_alphasift_bridge.py # 默认策略 python3 mo_alphasift_bridge.py --dry-run # 只看不写 python3 mo_alphasift_bridge.py --min-score 6 # 最低评分 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_MARKET = "cn" DEFAULT_MAX = 15 MIN_SCORE = 5 MAX_ADD = 5 POLL_INTERVAL = 5 POLL_TIMEOUT = 300 def load_json(path): try: return json.loads(path.read_text(encoding="utf-8")) 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 req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) except Exception as 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: continue key = "stocks" if "watchlist" in str(path) else "holdings" for item in data.get(key, []): c = str(item.get("code", "")).strip() if c: codes.add(c) return codes def run_screen(strategy, market, max_results, dry_run=False): now = datetime.now() date_str = now.strftime("%Y-%m-%d") time_str = now.strftime("%Y-%m-%d %H:%M") # 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: 提交任务失败") return task_id = task["task_id"] print(f" 任务ID: {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: continue s = status.get("status", "") if s == "completed": print(f" 完成 ({waited}s)") result = status.get("result", {}) break elif s == "failed": print(f" FAIL: {status.get('error', '')}") return else: pct = status.get("progress", 0) print(f" ...{s} ({pct}%)", flush=True) else: print(f" FAIL: 超时 ({POLL_TIMEOUT}s)") return candidates = result.get("candidates", []) if not candidates: print(" 无候选股") return 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 for c in candidates: code = str(c.get("code", "")).strip() score = c.get("score", 0) or c.get("llm_score", 0) or 0 name = c.get("name", "") or c.get("title", "") or code reason = c.get("reason", "") or c.get("llm_thesis", "") catalysts = c.get("llm_catalysts", "") risks = c.get("llm_risks", "") if score < MIN_SCORE: skipped += 1 continue if code in existing: continue notes_parts = [f"AlphaSift/{strategy} 评分{score}"] if reason: notes_parts.append(reason[:150]) if catalysts: notes_parts.append(f"催化剂:{catalysts[:100]}") new_stocks.append({ "code": code, "name": name, "price": c.get("price", 0), "source": "alpha_sift", "source_detail": { "strategy": strategy, "score": score, "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), "added_at": time_str, "added_by": "AlphaSift", "analysis": {}, }) existing.add(code) print(f" 过滤: {len(new_stocks)} 新标的 (跳过{skipped}只评分不足)") if not new_stocks: print(" 无符合条件的新标的") return new_stocks = new_stocks[:MAX_ADD] print(f"\n 新增 {len(new_stocks)} 只到自选池:") for s in new_stocks: print(f" {s['code']} {s['name']} (评分{s['source_detail']['score']})") if s['source_detail'].get('reason'): print(f" {s['source_detail']['reason'][:120]}") if 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}") # 5. 策略生成 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: 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" 股票已加入自选池，下次 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]}") def main(): global MIN_SCORE parser = argparse.ArgumentParser(description="AlphaSift → MoFin") parser.add_argument("--strategy", default=DEFAULT_STRATEGY) parser.add_argument("--market", default=DEFAULT_MARKET) parser.add_argument("--max", type=int, default=DEFAULT_MAX) parser.add_argument("--min-score", type=int, default=MIN_SCORE) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() MIN_SCORE = args.min_score if args.strategy == "list": list_strategies() return run_screen(args.strategy, args.market, args.max, args.dry_run) if __name__ == "__main__": main()