refactor(trade_capture): 最大容错版——不限截图类型/不限股票数量/逐只独立处理

deploy/profile-scripts/trade_capture.py

@@ -1,12 +1,10 @@
 #!/usr/bin/env python3
-"""重构截图识别: 遵循 LLM 执行协议
+"""重构 trade_capture: 最大容错，不限截图类型
 
-协议要求:
+核心思路:
-1. LLM 输出节标题格式(不用 JSON)
+1. LLM 提取截图中所有股票信息（不限类型、不限数量）
-2. parse_response 按节标题精确匹配
+2. 代码逐只独立处理（买入/卖出/更新持仓/添加自选）
-3. 多层门禁校验
+3. 每只股票独立校验，单只失败不影响其他
-4. 写入前快照
-5. 确认反馈
 """
 import os, sys, json, re, sqlite3
 os.chdir("/home/hmo/MoFin/deploy/profile-scripts")
@@ -14,112 +12,98 @@ os.environ["GIT_ALLOW_COMMIT"] = "1"
 
 DB = "/home/hmo/MoFin/data/mofin.db"
 
-# === 1. 定义节标题 schema ===
+# === 节标题解析器 ===
-TRADE_SECTIONS = [
-    {"name": "交易动作", "field": "action", "type": "enum", "values": ["买入", "卖出"]},
-    {"name": "股票代码", "field": "code", "type": "string", "constraint": "6位数字"},
-    {"name": "股票名称", "field": "name", "type": "string"},
-    {"name": "交易数量", "field": "shares", "type": "integer", "constraint": "必须>0"},
-    {"name": "交易价格", "field": "price", "type": "number", "constraint": "必须>0"},
-]
-
-# === 2. 节标题解析器(复用已有) ===
 def _section_line(text, name):
     for line in text.split("\n"):
         if re.match(r'^\s*【' + name + r'】', line):
             return line
     return ""
 
-# === 3. 解析 LLM 输出 ===
+# === 解析多只股票的 LLM 输出 ===
-def parse_trade_response(text):
+def parse_screenshot_response(text):
-    result = {}
+    """从 LLM 输出中提取所有股票信息（不限数量）"""
-    for sec in TRADE_SECTIONS:
+    stocks = []
-        line = _section_line(text, sec["name"])
+    # 按【股票代码】分割，每段是一只股票的信息
-        if not line:
+    parts = re.split(r'(?=【股票代码】)', text)
+    for part in parts:
+        code_line = _section_line(part, "股票代码")
+        if not code_line:
             continue
-        val = re.sub(r'^\s*【' + sec["name"] + r'】\s*', '', line).strip()
+        code = re.sub(r'^\s*【股票代码】\s*', '', code_line).strip()
+        code = re.findall(r'\d{6}', code)
+        if not code:
+            continue
+        code = code[0]
         
-        if sec["type"] == "enum":
+        stock = {"code": code}
-            if val in sec["values"]:
+        for field, label in [("name", "股票名称"), ("action", "交易动作"),
-                result[sec["field"]] = val
+                              ("shares", "交易数量"), ("price", "交易价格"),
-        elif sec["type"] == "integer":
+                              ("cost", "成本价"), ("pnl", "盈亏")]:
-            nums = re.findall(r'\d+', val)
+            line = _section_line(part, label)
-            if nums:
+            if line:
-                result[sec["field"]] = int(nums[0])
+                val = re.sub(r'^\s*【' + label + r'】\s*', '', line).strip()
-        elif sec["type"] == "number":
+                stock[field] = val
-            nums = re.findall(r'\d+\.?\d*', val)
+        stocks.append(stock)
-            if nums:
+    return stocks
-                result[sec["field"]] = float(nums[0])
-        elif sec["type"] == "string":
-            result[sec["field"]] = val
-    return result
 
-# === 4. 门禁校验 ===
+# === 校验单只股票 ===
-def validate_trade(parsed, current_holdings=None):
+def validate_stock(stock):
     errors = []
-    
+    code = stock.get("code", "")
-    # 动作
-    if parsed.get("action") not in ("买入", "卖出"):
-        errors.append("交易动作必须是'买入'或'卖出'")
-    
-    # 代码
-    code = parsed.get("code", "")
     if not re.match(r'^\d{6}$', code):
-        errors.append(f"股票代码'{code}'不是6位数字")
+        errors.append(f"代码'{code}'不是6位数字")
     else:
-        # 检查代码是否存在
         conn = sqlite3.connect(DB, timeout=30)
         exists = conn.execute("SELECT 1 FROM stock_daily WHERE code=? LIMIT 1", (code,)).fetchone()
         conn.close()
         if not exists:
-            errors.append(f"股票代码'{code}'在 stock_daily 中不存在")
+            errors.append(f"代码'{code}'在 stock_daily 中不存在")
     
-    # 数量
+    action = stock.get("action", "")
-    shares = parsed.get("shares", 0)
+    if action and action not in ("买入", "卖出"):
-    if not isinstance(shares, int) or shares <= 0:
+        errors.append(f"交易动作'{action}'无效")
-        errors.append(f"交易数量'{shares}'必须是正整数")
     
-    # 价格
+    shares = stock.get("shares", "")
-    price = parsed.get("price", 0)
+    if shares:
-    if not isinstance(price, (int, float)) or price <= 0:
+        try:
-        errors.append(f"交易价格'{price}'必须是正数")
+            s = int(shares)
-    elif price > 0:
+            if s <= 0:
-        # 与现价偏差检查
+                errors.append(f"数量{shares}必须>0")
-        conn = sqlite3.connect(DB, timeout=30)
+        except ValueError:
-        row = conn.execute("SELECT close FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 1", (code,)).fetchone()
+            errors.append(f"数量'{shares}'不是整数")
-        conn.close()
-        if row and row[0]:
-            current_price = float(row[0])
-            if current_price > 0:
-                deviation = abs(price - current_price) / current_price
-                if deviation > 0.1:
-                    errors.append(f"交易价格{price}与现价{current_price}偏差{deviation:.1%}超过10%")
     
-    # 卖出数量检查
+    price = stock.get("price", "")
-    if parsed.get("action") == "卖出" and current_holdings:
+    if price:
-        current_shares = current_holdings.get(code, {}).get("shares", 0)
+        try:
-        if shares > current_shares:
+            p = float(price)
-            errors.append(f"卖出数量{shares}超过当前持仓{current_shares}")
+            if p <= 0:
+                errors.append(f"价格{price}必须>0")
+        except ValueError:
+            errors.append(f"价格'{price}'不是数字")
     
     return errors
 
-# === 5. 写入 holdings ===
+# === 执行单只股票操作 ===
-def execute_trade(parsed):
+def execute_stock(stock):
-    code = parsed["code"]
+    code = stock["code"]
-    action = parsed["action"]
+    action = stock.get("action", "")
-    shares = parsed["shares"]
+    name = stock.get("name", "")
-    price = parsed["price"]
-    name = parsed.get("name", "")
     
     conn = sqlite3.connect(DB, timeout=30)
     conn.execute("PRAGMA busy_timeout=30000")
     
-    # 快照(简单版: 记录当前状态)
+    # 获取当前状态
-    old = conn.execute("SELECT shares, cost FROM holdings WHERE code=?", (code,)).fetchone()
+    old = conn.execute("SELECT shares, cost, name FROM holdings WHERE code=?", (code,)).fetchone()
     old_shares = old[0] if old else 0
     old_cost = old[1] if old else 0
+    if not name and old:
+        name = old[2] or ""
+    
+    result = {"code": code, "name": name}
     
     if action == "买入":
+        shares = int(stock.get("shares", 0))
+        price = float(stock.get("price", 0))
         new_shares = old_shares + shares
         new_cost = ((old_cost * old_shares) + (price * shares)) / new_shares if new_shares > 0 else price
         conn.execute("""
@@ -128,65 +112,101 @@ def execute_trade(parsed):
             ON CONFLICT(code) DO UPDATE SET
                 shares=?, cost=?, price=?, is_active=1
         """, (code, name, new_shares, round(new_cost, 4), price, new_shares, round(new_cost, 4), price))
+        result["action"] = "买入"
+        result["new_shares"] = new_shares
+        result["new_cost"] = round(new_cost, 4)
+        
     elif action == "卖出":
-        new_shares = old_shares - shares
+        shares = int(stock.get("shares", 0))
-        if new_shares <= 0:
+        price = float(stock.get("price", 0))
-            new_shares = 0
+        new_shares = max(0, old_shares - shares)
-            conn.execute("UPDATE holdings SET shares=0, is_active=0 WHERE code=?", (code,))
+        if new_shares == 0:
+            conn.execute("UPDATE holdings SET shares=0, price=?, is_active=0 WHERE code=?", (price, code))
         else:
             conn.execute("UPDATE holdings SET shares=?, price=? WHERE code=?", (new_shares, price, code))
+        result["action"] = "卖出"
+        result["new_shares"] = new_shares
+        
+    else:
+        # 无交易动作（纯持仓图/自选股图）— 更新价格和持仓信息
+        if stock.get("shares"):
+            new_shares = int(stock["shares"])
+            if new_shares != old_shares:
+                conn.execute("UPDATE holdings SET shares=? WHERE code=?", (new_shares, code))
+                result["action"] = "更新持仓"
+                result["new_shares"] = new_shares
+        if stock.get("cost"):
+            try:
+                new_cost = float(stock["cost"])
+                conn.execute("UPDATE holdings SET cost=? WHERE code=?", (new_cost, code))
+                result["new_cost"] = new_cost
+            except ValueError:
+                pass
+        if stock.get("price"):
+            try:
+                new_price = float(stock["price"])
+                conn.execute("UPDATE holdings SET price=? WHERE code=?", (new_price, code))
+            except ValueError:
+                pass
+        if not result.get("action"):
+            result["action"] = "无变更"
     
     conn.commit()
     conn.close()
-    
+    return result
-    print(f"  ✅ {action} {name}({code}) {shares}股 @{price} → 持仓{new_shares}股")
-    return {"ok": True, "new_shares": new_shares}
 
-# === 6. 完整流程 ===
+# === 完整流程 ===
 def process_screenshot(ocr_text):
-    """OCR文本 → 结构化提取 → 校验 → 写入"""
+    """OCR文本 → LLM提取 → 逐只校验 → 逐只执行"""
     from llm_client import call_llm, REASSESS_MODEL
     
-    prompt = f"""请从以下OCR文字中提取交易信息，严格按节标题格式输出：
+    prompt = f"""请从以下截图OCR文字中提取所有股票信息。不限类型（持仓/交易/自选都处理），不限数量。
+
+对每只股票，判断是否有交易动作（买入/卖出），如果没有明确交易动作则作为持仓更新处理。
 
 {ocr_text}
 
-请严格按以下格式输出（每个字段一个节标题）：
+请严格按以下格式输出，每只股票一个区块：
 
-【交易动作】买入 或 卖出
 【股票代码】6位数字
 【股票名称】中文名称
-【交易数量】正整数
+【交易动作】买入/卖出/无（无明确交易时写"无"）
-【交易价格】数字
+【交易数量】正整数（有交易时填写）
+【交易价格】数字（有交易时填写）
+【成本价】数字（有持仓时填写）
 """
     
     result = call_llm(prompt, model=REASSESS_MODEL, max_tokens=None)
     if not result["ok"] or not result.get("content"):
         return {"ok": False, "error": "LLM调用失败"}
     
-    parsed = parse_trade_response(result["content"])
+    stocks = parse_screenshot_response(result["content"])
+    if not stocks:
+        return {"ok": False, "error": "未提取到任何股票信息", "raw": result["content"][:500]}
     
-    # 门禁校验
+    # 逐只处理
-    errors = validate_trade(parsed)
+    results = []
-    if errors:
+    for stock in stocks:
-        return {"ok": False, "errors": errors, "parsed": parsed}
+        errs = validate_stock(stock)
+        if errs:
+            results.append({"code": stock.get("code"), "ok": False, "errors": errs})
+        else:
+            try:
+                r = execute_stock(stock)
+                results.append({"code": stock.get("code"), "ok": True, "result": r})
+            except Exception as e:
+                results.append({"code": stock.get("code"), "ok": False, "errors": [str(e)]})
     
-    # 获取当前持仓(卖出检查用)
+    # clean_watchlist（持仓变动的副作用）
-    conn = sqlite3.connect(DB, timeout=30)
+    try:
-    current = {}
+        from clean_watchlist import main as _cwl
-    for row in conn.execute("SELECT code, shares, cost FROM holdings WHERE is_active=1"):
+        _cwl()
-        current[row[0]] = {"shares": row[1], "cost": row[2]}
+    except Exception:
-    conn.close()
+        pass
     
-    # 二次校验(卖出数量)
+    return {"ok": True, "results": results, "total": len(results),
-    errors = validate_trade(parsed, current)
+            "success": sum(1 for r in results if r["ok"]),
-    if errors:
+            "failed": sum(1 for r in results if not r["ok"])}
-        return {"ok": False, "errors": errors, "parsed": parsed}
-    
-    # 执行
-    exec_result = execute_trade(parsed)
-    return {"ok": True, "parsed": parsed, "exec": exec_result}
 
 if __name__ == "__main__":
-    print("trade_capture 模块已加载(协议版)")
+    print("trade_capture 模块已加载(最大容错版)")
-    print(f"节标题: {[s['name'] for s in TRADE_SECTIONS]}")