#!/usr/bin/env python3 """batch_reassess.py — 批量补全九维分析(逐只处理,间隔防限流) 用法: python3 batch_reassess.py [--all] [--code XXXXXX] 流程:收集最新数据 → 调LLM(gateway)写九维分析+策略 → 保存到DB """ import sys, json, subprocess, sqlite3, re, time from datetime import datetime DB = "/home/hmo/MoFin/data/mofin.db" GATEWAY = "http://127.0.0.1:8643/v1/chat/completions" COOLDOWN_HOURS = 1 def has_llm_analysis(code): """检查是否为LLM生成的九维分析(>500字)""" conn = sqlite3.connect(DB) r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() conn.close() return r and r[0] and r[0] > 500 def in_cooldown(code): """冷却期检查""" conn = sqlite3.connect(DB) r = conn.execute("SELECT reassessed_at FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() conn.close() if not r or not r[0]: return False try: last = datetime.fromisoformat(r[0]) diff = (datetime.now() - last).total_seconds() / 3600 return diff < COOLDOWN_HOURS except: return False def collect_data(code): """收集最新数据""" data = {"code": code} # 从DB读策略 conn = sqlite3.connect(DB) r = conn.execute("SELECT name, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, rr_ratio, tech_snapshot, sector_context, stock_category FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if r: data["name"] = r[0] data["entry_low"] = r[1] or 0 data["entry_high"] = r[2] or 0 data["stop_loss"] = r[3] or 0 data["take_profit"] = r[4] or 0 data["timing_signal"] = r[5] or "" data["action"] = r[6] or "" data["rr_ratio"] = r[7] or 0 data["tech_snapshot"] = r[8] or "" data["sector_context"] = r[9] or "" data["stock_category"] = r[10] or "" conn.close() # 从腾讯API拉最新价和基本面 prefix = "sh" if str(code).startswith(("6","9")) else "sz" try: r = subprocess.run(["curl", "-s", f"http://qt.gtimg.cn/q={prefix}{code}"], capture_output=True, timeout=10) parts = r.stdout.decode("gbk", errors="ignore").split("~") data["price"] = float(parts[3]) if len(parts) > 3 and parts[3] else 0 data["pe"] = parts[39] if len(parts) > 39 and parts[39] else "" data["mcap"] = parts[44] if len(parts) > 44 and parts[44] else "" data["change_pct"] = parts[32] if len(parts) > 32 and parts[32] else "0" except: data["price"] = 0 # 大盘 try: conn = sqlite3.connect(DB) mr = conn.execute("SELECT structure FROM macro_context_log ORDER BY id DESC LIMIT 1").fetchone() if mr and mr[0]: s = json.loads(mr[0]) data["macro"] = s.get("description", "大盘震荡") conn.close() except: data["macro"] = "大盘震荡" return data def build_prompt(data): """构建LLM prompt,要求输出完整策略""" return f"""你是一个资深A股分析师。请对{data['code']} {data.get('name','')}做一个完整的九维矩阵分析,并输出策略参数。 当前数据: 大盘:{data.get('macro','震荡')} 最新价:{data.get('price',0)} 涨跌:{data.get('change_pct','0')}% PE={data.get('pe','?')} 市值={data.get('mcap','?')}亿 行业:{data.get('sector_context','?')} 技术面:{data.get('tech_snapshot','')[:200]} 当前信号:{data.get('timing_signal','?')} 分类:{data.get('stock_category','?')} 原策略:{(data.get('action','') or '')[:200]} 请严格按以下格式输出: ① 大盘×基本面 [一句话] ② 大盘×消息面 [一句话] ③ 大盘×技术面 [一句话] ④ 大盘×资金流 [一句话] ⑤ 行业×基本面 [一句话] ⑥ 行业×消息面 [一句话] ⑦ 行业×技术面 [一句话] ⑧ 个股×基本面 [一句话] ⑨ 个股×消息面 [一句话] 【综合结论】(买入/关注/观望/卖出) 【操作建议】具体操作建议 【买入区间】最低价~最高价 【建议止损】数字 【建议止盈】数字 【建议仓位】总资产的百分之几(如8%),只写数字+%号,不要写文字描述""" def parse_response(text): """从LLM回复中提取策略参数""" result = {"signal": "", "entry_low": 0, "entry_high": 0, "stop_loss": 0, "take_profit": 0, "position": ""} # 信号 sl = [l for l in text.split("\n") if "综合结论" in l] if sl: for kw in ["买入","关注","观望","卖出"]: if kw in sl[0]: result["signal"] = kw break # 买入区间 zl = [l for l in text.split("\n") if "买入区间" in l] if zl: nums = re.findall(r'[\d.]+', zl[0]) if len(nums) >= 2: result["entry_low"] = float(nums[0]) result["entry_high"] = float(nums[1]) # 止损 for l in text.split("\n"): if "建议止损" in l: nums = re.findall(r'[\d.]+', l) if nums: result["stop_loss"] = float(nums[0]) # 止盈 for l in text.split("\n"): if "建议止盈" in l: nums = re.findall(r'[\d.]+', l) if nums: result["take_profit"] = float(nums[0]) # 仓位(提取百分比) for l in text.split("\n"): if "建议仓位" in l: nums = re.findall(r'[\d.]+', l) pct = "" if nums: # 取第一个合理的百分比(1-100之间) for n in nums: f = float(n) if 1 <= f <= 100: pct = f"{f:.0f}%" break # 如果没有百分比,用文字描述映射到百分比 raw = l.replace("建议仓位","").strip() if not pct: if "轻仓" in raw: pct = "3%" elif "中" in raw and "仓" in raw: pct = "5%" elif "重仓" in raw: pct = "10%" elif "清仓" in raw or "零仓" in raw: pct = "0%" else: pct = "5%" result["position"] = pct break return result def save_result(code, full_text, parsed): """保存LLM结果到DB""" conn = sqlite3.connect(DB) now = datetime.now().isoformat() updates = ["full_analysis=?", "reassessed_at=?"] params = [full_text, now] if parsed["signal"]: updates.append("timing_signal=?") params.append(parsed["signal"]) if parsed["entry_low"] > 0: updates.append("entry_low=?") params.append(parsed["entry_low"]) if parsed["entry_high"] > 0: updates.append("entry_high=?") params.append(parsed["entry_high"]) if parsed["stop_loss"] > 0: updates.append("stop_loss=?") params.append(parsed["stop_loss"]) if parsed["take_profit"] > 0: updates.append("take_profit=?") params.append(parsed["take_profit"]) if parsed["position"]: updates.append("position_advice=?") params.append(parsed["position"]) params.append(code) sql = f"UPDATE holding_strategies SET {', '.join(updates)} WHERE code=? AND status='active'" conn.execute(sql, params) conn.commit() conn.close() def process_stock(code): """处理单只股票""" print(f"\n{'='*50}") print(f"处理: {code}") print(f"{'='*50}") if has_llm_analysis(code): print(f" ⏭ 已有LLM九维分析,跳过") return False if in_cooldown(code): print(f" ⏭ 冷却期内,跳过") return False print(f" 收集数据...", flush=True) data = collect_data(code) if not data.get("price"): print(f" ⚠️ 无价格数据,跳过") return False print(f" 调LLM生成九维分析...", flush=True) prompt = build_prompt(data) try: r = subprocess.run(["curl", "-s", "--max-time", "300", "-H", "Content-Type: application/json", "-H", "Authorization: Bearer hermes123", "-d", json.dumps({"model":"deepseek-v4-flash","messages":[{"role":"user","content":prompt}],"max_tokens":2048}), GATEWAY], capture_output=True, timeout=310) if r.returncode != 0: print(f" ❌ curl失败: {r.stderr.decode()[:100]}") return False resp = json.loads(r.stdout) if "choices" not in resp: print(f" ❌ API异常: {str(resp)[:200]}") return False full_text = resp["choices"][0]["message"]["content"] print(f" ✅ LLM返回({len(full_text)}字)", flush=True) parsed = parse_response(full_text) print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}") save_result(code, full_text, parsed) print(f" ✅ 已保存到DB") return True except subprocess.TimeoutExpired: print(f" ❌ 超时") return False except Exception as e: print(f" ❌ 错误: {e}") return False def main(): codes = [] if "--code" in sys.argv: idx = sys.argv.index("--code") codes = [sys.argv[idx+1]] else: # 所有自选策略 conn = sqlite3.connect(DB) rows = conn.execute("SELECT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略' ORDER BY code").fetchall() conn.close() codes = [r[0] for r in rows] print(f"待处理: {len(codes)}只") ok = 0 fail = 0 skip = 0 for i, code in enumerate(codes): if has_llm_analysis(code): print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有LLM分析") skip += 1 continue print(f" [{i+1}/{len(codes)}] ", end="", flush=True) if process_stock(code): ok += 1 else: fail += 1 # 间隔15秒(防gateway过载) if i < len(codes) - 1: print(f" 等待15秒...", flush=True) time.sleep(15) print(f"\n{'='*50}") print(f"完成: {ok}成功, {fail}失败, {skip}跳过") print(f"{'='*50}") if __name__ == "__main__": main()