From a2ef70f60750ab7aa94d2b979252935db0b15cc2 Mon Sep 17 00:00:00 2001
From: Lxy <windsdreaming@163.com>
Date: Mon, 6 Jul 2026 00:47:24 +0800
Subject: [PATCH] test: add trade reflection unit/integration tests and fix
 pairing engine bugs

- Added tests/test_trade_reflection.py with 13 test cases:
  - FIFO pairing, cross-day pairing, multiple opens merge
  - Manual pair, delete pair keeps records
  - Daily trades view with reflection status
  - Tag deduplication, preset tags
  - AI prompt building, AI response parsing (JSON/Markdown/fallback)
  - Save AI suggestion as experience
- Fixed pairing engine:
  - Imported missing DailyReflection
  - Enabled cross-day pairing by including previous unpaired records
  - Merged multiple opens into single position per close
  - Fixed operator precedence in price/commission calculations
- All tests pass: 123 passed, 2 xfailed
---
 app/services/trade_pairing_engine.py | 138 ++++++++-----
 tests/test_trade_reflection.py       | 287 +++++++++++++++++++++++++++
 2 files changed, 374 insertions(+), 51 deletions(-)
 create mode 100644 tests/test_trade_reflection.py

diff --git a/app/services/trade_pairing_engine.py b/app/services/trade_pairing_engine.py
index 7eac820..900dc63 100644
--- a/app/services/trade_pairing_engine.py
+++ b/app/services/trade_pairing_engine.py
@@ -2,35 +2,54 @@
 交易配对引擎 - 将开仓和平仓配对为完整交易，持久化到 TradePair 表
 """
 import logging
-from datetime import datetime
 from typing import Optional
 
 from sqlalchemy.orm import Session
 from collections import defaultdict
 
-from app.analysis_models import TradeRecord, TradePair, TradeReflection
+from app.analysis_models import TradeRecord, TradePair, TradeReflection, DailyReflection
 
 logger = logging.getLogger(__name__)
 
 
 def auto_pair_trades(db: Session, trade_date: str) -> dict:
     """
-    自动配对指定日期的交易记录
+    自动配对指定日期及之前未平仓的交易记录
     按品种+方向分组，FIFO 规则配对
     返回：{created: 新建配对数, updated: 更新配对数, unpaired: 未配对记录数}
     """
-    records = db.query(TradeRecord).filter(
+    # 查询指定日期的记录
+    current_records = db.query(TradeRecord).filter(
         TradeRecord.trade_date == trade_date
     ).order_by(
         TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time
     ).all()
 
-    if not records:
+    if not current_records:
         return {"created": 0, "updated": 0, "unpaired": 0}
 
+    # 查询之前未配对的记录（跨日持仓）
+    # 找出所有已经配对的记录 ID
+    all_pairs = db.query(TradePair).all()
+    paired_ids = set()
+    for p in all_pairs:
+        paired_ids.update(p.open_record_ids or [])
+        paired_ids.update(p.close_record_ids or [])
+
+    # 查询 trade_date 之前的未配对记录
+    previous_unpaired = db.query(TradeRecord).filter(
+        TradeRecord.trade_date < trade_date,
+        TradeRecord.id.notin_(paired_ids) if paired_ids else True,
+    ).order_by(
+        TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time
+    ).all()
+
+    # 合并当前记录和之前未配对记录
+    all_records = previous_unpaired + current_records
+
     # 按品种分组
     by_variety = defaultdict(list)
-    for r in records:
+    for r in all_records:
         by_variety[r.variety].append(r)
 
     result = {"created": 0, "updated": 0, "unpaired": 0}
@@ -68,11 +87,10 @@ def _pair_single_variety(db: Session, variety: str, records: list) -> dict:
     all_paired_close_ids = set()
     existing_pairs = db.query(TradePair).filter(
         TradePair.variety == variety,
-        TradePair.open_date >= records[0].trade_date if records else None
     ).all()
     for p in existing_pairs:
-        all_paired_open_ids.update(p.open_record_ids)
-        all_paired_close_ids.update(p.close_record_ids)
+        all_paired_open_ids.update(p.open_record_ids or [])
+        all_paired_close_ids.update(p.close_record_ids or [])
 
     unpaired_count = 0
     for r in records:
@@ -84,59 +102,77 @@ def _pair_single_variety(db: Session, variety: str, records: list) -> dict:
 
 
 def _pair_direction(db: Session, variety: str, opens: list, closes: list, direction: str) -> int:
-    """单方向配对（FIFO），返回新建配对数"""
+    """单方向配对（FIFO，支持多开合并），返回新建配对数
+    
+    每个平仓记录会尽可能与前面的开仓记录配对，形成一个完整配对。
+    如果多个开仓合并才能匹配一个平仓，则这些开仓记录会归入同一个配对。
+    """
     created = 0
-    open_queue = list(opens)  # FIFO 队列
-    close_queue = list(closes)
+    # 开仓队列，每个元素是 (record, remaining_volume)
+    open_queue = [(r, r.volume or 0) for r in opens]
+    close_queue = [(r, r.volume or 0) for r in closes]
 
     while open_queue and close_queue:
-        open_rec = open_queue[0]
-        close_rec = close_queue[0]
-
-        open_volume = open_rec.volume or 0
-        close_volume = close_rec.volume or 0
-
-        # 计算配对手数
-        pair_volume = min(open_volume, close_volume)
-
-        # 计算平均价格
-        open_price = open_rec.price
-        close_price = close_rec.price
-
-        # 计算盈亏
-        close_pnl = close_rec.close_pnl or 0
-        total_commission = (open_rec.commission or 0) + (close_rec.commission or 0)
+        close_rec, close_remaining = close_queue[0]
+
+        # 从开仓队列中取足够手数来匹配这个平仓
+        needed = close_remaining
+        used_opens = []  # (record, take_volume)
+        while needed > 0 and open_queue:
+            open_rec, open_remaining = open_queue[0]
+            take = min(open_remaining, needed)
+            used_opens.append((open_rec, take))
+            needed -= take
+            if take >= open_remaining:
+                open_queue.pop(0)
+            else:
+                open_queue[0] = (open_rec, open_remaining - take)
+
+        if not used_opens:
+            break
+
+        # 实际匹配的手数
+        matched_volume = close_remaining - needed
+
+        # 更新平仓队列
+        if matched_volume >= close_remaining:
+            close_queue.pop(0)
+        else:
+            close_queue[0] = (close_rec, close_remaining - matched_volume)
+
+        # 计算均价（按手数加权）
+        open_volume_sum = sum(take for _, take in used_opens)
+        open_price = sum((r.price or 0) * take for r, take in used_opens) / open_volume_sum if open_volume_sum > 0 else 0
+        close_price = close_rec.price or 0
+
+        # 计算盈亏和手续费
+        # 平仓盈亏按比例分配到该平仓记录
+        close_pnl = (close_rec.close_pnl or 0) * (matched_volume / close_rec.volume) if close_rec.volume else 0
+        total_commission = sum((r.commission or 0) * (take / r.volume) for r, take in used_opens if r.volume) + \
+                          (close_rec.commission or 0) * (matched_volume / close_rec.volume) if close_rec.volume else 0
         net_pnl = close_pnl - total_commission
 
-        # 创建配对记录
+        # 最早开仓日期和最早平仓日期
+        open_date = min(r.trade_date for r, _ in used_opens)
+        close_date = close_rec.trade_date
+
         pair = TradePair(
             variety=variety,
             direction=direction,
-            open_record_ids=[open_rec.id],
+            open_record_ids=[r.id for r, _ in used_opens],
             close_record_ids=[close_rec.id],
-            open_price=open_price,
-            close_price=close_price,
-            total_volume=pair_volume,
-            close_pnl=close_pnl,
-            total_commission=total_commission,
-            net_pnl=net_pnl,
-            open_date=open_rec.trade_date,
-            close_date=close_rec.trade_date,
+            open_price=round(open_price, 2),
+            close_price=round(close_price, 2),
+            total_volume=matched_volume,
+            close_pnl=round(close_pnl, 2),
+            total_commission=round(total_commission, 2),
+            net_pnl=round(net_pnl, 2),
+            open_date=open_date,
+            close_date=close_date,
         )
         db.add(pair)
         created += 1
 
-        # 从队列中移除或更新剩余手数
-        if open_volume == pair_volume:
-            open_queue.pop(0)
-        else:
-            open_rec.volume = open_volume - pair_volume
-
-        if close_volume == pair_volume:
-            close_queue.pop(0)
-        else:
-            close_rec.volume = close_volume - pair_volume
-
     db.flush()
     return created
 
@@ -200,8 +236,8 @@ def get_daily_trades_with_pairs(db: Session, trade_date: str) -> dict:
     # 未配对记录
     paired_record_ids = set()
     for p in pairs:
-        paired_record_ids.update(p.open_record_ids)
-        paired_record_ids.update(p.close_record_ids)
+        paired_record_ids.update(p.open_record_ids or [])
+        paired_record_ids.update(p.close_record_ids or [])
 
     unpaired = [r for r in records if r.id not in paired_record_ids]
     unpaired_data = [{
diff --git a/tests/test_trade_reflection.py b/tests/test_trade_reflection.py
new file mode 100644
index 0000000..5d53da6
--- /dev/null
+++ b/tests/test_trade_reflection.py
@@ -0,0 +1,287 @@
+"""
+交易反思系统单元测试与集成测试
+"""
+import json
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.analysis_db import AnalysisBase
+from app.analysis_models import (
+    TradeRecord, TradePair, TradeReflection, DailyReflection,
+    TradeTag, TradeRecordTag, TradeExperience, TradeAIAnalysis,
+)
+from app.services.trade_pairing_engine import (
+    auto_pair_trades,
+    create_manual_pair,
+    delete_pair,
+    get_daily_trades_with_pairs,
+)
+from app.services.reflection_ai_analysis import (
+    build_reflection_prompt,
+    _parse_ai_response,
+    save_suggestion_as_experience,
+)
+
+
+@pytest.fixture
+def db():
+    """提供已创建表结构的内存 SQLite 会话（analysis 库）。"""
+    engine = create_engine("sqlite:///:memory:")
+    AnalysisBase.metadata.create_all(bind=engine)
+    Session = sessionmaker(bind=engine)
+    session = Session()
+    yield session
+    session.close()
+
+
+def _make_record(db, symbol="AG2608", variety="AG", direction="买", offset="开",
+                 price=7850, volume=2, close_pnl=0, commission=10,
+                 trade_date="2026-07-05", trade_time="09:30:00", import_batch="test-batch"):
+    """辅助函数：创建交易记录"""
+    r = TradeRecord(
+        trade_type="期货",
+        symbol=symbol,
+        variety=variety,
+        symbol_name="沪银",
+        direction=direction,
+        offset=offset,
+        price=price,
+        volume=volume,
+        close_pnl=close_pnl,
+        commission=commission,
+        trade_date=trade_date,
+        trade_time=trade_time,
+        import_batch=import_batch,
+    )
+    db.add(r)
+    db.flush()
+    return r
+
+
+class TestTradePairingEngine:
+    """交易配对引擎测试"""
+
+    def test_fifo_pairing_single_day(self, db):
+        """单日 FIFO 配对：先开先平"""
+        open1 = _make_record(db, direction="买", offset="开", price=7850, volume=2, trade_time="09:30:00")
+        close1 = _make_record(db, direction="卖", offset="平", price=7910, volume=2,
+                              close_pnl=1200, commission=10, trade_time="14:20:00")
+
+        result = auto_pair_trades(db, "2026-07-05")
+        db.commit()
+
+        assert result["created"] == 1
+        pairs = db.query(TradePair).all()
+        assert len(pairs) == 1
+        assert pairs[0].direction == "long"
+        assert pairs[0].open_price == 7850
+        assert pairs[0].close_price == 7910
+        assert pairs[0].net_pnl == 1180  # 1200 - 10 - 10
+
+    def test_cross_day_pairing(self, db):
+        """跨日配对：开仓和平仓在不同日期"""
+        open_rec = _make_record(db, direction="买", offset="开", price=4000, volume=1,
+                                trade_date="2026-07-04", trade_time="09:30:00")
+        close_rec = _make_record(db, direction="卖", offset="平", price=4100, volume=1,
+                                 close_pnl=1000, commission=20,
+                                 trade_date="2026-07-05", trade_time="14:00:00")
+
+        # 先配对 7-04 的数据（只有开仓，不会配对）
+        auto_pair_trades(db, "2026-07-04")
+        # 再配对 7-05 的数据
+        result = auto_pair_trades(db, "2026-07-05")
+        db.commit()
+
+        pairs = db.query(TradePair).all()
+        assert len(pairs) == 1
+        assert pairs[0].open_date == "2026-07-04"
+        assert pairs[0].close_date == "2026-07-05"
+        assert pairs[0].net_pnl == 970  # 1000 - 10 - 20
+
+    def test_multiple_opens_merged_position(self, db):
+        """多次开仓合并为一个头寸，FIFO 配对"""
+        open1 = _make_record(db, direction="买", offset="开", price=7800, volume=1, trade_time="09:30:00")
+        open2 = _make_record(db, direction="买", offset="开", price=7820, volume=1, trade_time="10:00:00")
+        close1 = _make_record(db, direction="卖", offset="平", price=7900, volume=2,
+                              close_pnl=1800, commission=20, trade_time="14:00:00")
+
+        result = auto_pair_trades(db, "2026-07-05")
+        db.commit()
+
+        assert result["created"] == 1
+        pairs = db.query(TradePair).all()
+        assert len(pairs) == 1
+        # 由于 close_volume(2) == open1.volume(1) + open2.volume(1)，会一起配对
+        assert pairs[0].total_volume == 2
+
+    def test_manual_pair(self, db):
+        """手动创建配对"""
+        open_rec = _make_record(db, direction="买", offset="开", price=7800, volume=2)
+        close_rec = _make_record(db, direction="卖", offset="平", price=7900, volume=2,
+                                 close_pnl=2000, commission=20)
+
+        pair = create_manual_pair(db, [open_rec.id], [close_rec.id])
+        db.commit()
+
+        assert pair.direction == "long"
+        assert pair.open_price == 7800
+        assert pair.close_price == 7900
+        assert pair.net_pnl == 1970  # 2000 - 10 - 20
+
+    def test_delete_pair_keeps_records(self, db):
+        """删除配对不删除原始交易记录"""
+        open_rec = _make_record(db, direction="买", offset="开", price=7800, volume=1)
+        close_rec = _make_record(db, direction="卖", offset="平", price=7900, volume=1,
+                                 close_pnl=1000, commission=10)
+        pair = create_manual_pair(db, [open_rec.id], [close_rec.id])
+        db.commit()
+
+        delete_pair(db, pair.id)
+        db.commit()
+
+        assert db.query(TradePair).count() == 0
+        assert db.query(TradeRecord).count() == 2
+
+
+class TestDailyTradesView:
+    """按天交易视图测试"""
+
+    def test_get_daily_trades_with_pairs(self, db):
+        """获取某日交易+配对+反思状态"""
+        open_rec = _make_record(db, direction="买", offset="开", price=7800, volume=1)
+        close_rec = _make_record(db, direction="卖", offset="平", price=7900, volume=1,
+                                 close_pnl=1000, commission=10)
+        # 创建反思
+        pair = create_manual_pair(db, [open_rec.id], [close_rec.id])
+        reflection = TradeReflection(
+            trade_pair_id=pair.id,
+            trade_date="2026-07-05",
+            entry_reason="测试",
+            discipline_score=4,
+        )
+        db.add(reflection)
+        db.commit()
+
+        data = get_daily_trades_with_pairs(db, "2026-07-05")
+
+        assert data["stats"]["paired_count"] == 1
+        assert data["stats"]["unpaired_count"] == 0
+        assert len(data["pairs"]) == 1
+        assert data["pairs"][0]["reflection_status"] == "done"
+
+
+class TestTagSystem:
+    """标签系统测试"""
+
+    def test_tag_deduplication(self, db):
+        """同一交易不能重复打相同标签"""
+        tag = TradeTag(name="追涨杀跌", category="operation", is_preset=True)
+        db.add(tag)
+        db.flush()
+
+        relation1 = TradeRecordTag(trade_pair_id=1, tag_id=tag.id)
+        db.add(relation1)
+        db.commit()
+
+        # 重复添加应该通过唯一约束失败
+        relation2 = TradeRecordTag(trade_pair_id=1, tag_id=tag.id)
+        db.add(relation2)
+        with pytest.raises(Exception):
+            db.commit()
+        db.rollback()
+
+    def test_preset_tags_initialization(self, db):
+        """预设标签可以正常创建"""
+        tags = [
+            TradeTag(name="严格执行计划", category="discipline", is_preset=True),
+            TradeTag(name="突破交易", category="technical", is_preset=True),
+        ]
+        db.add_all(tags)
+        db.commit()
+
+        assert db.query(TradeTag).filter_by(is_preset=True).count() == 2
+
+
+class TestReflectionAIPrompt:
+    """AI 反思分析提示词测试"""
+
+    def test_build_reflection_prompt_contains_key_info(self, db):
+        """提示词包含交易关键信息和反思内容"""
+        open_rec = _make_record(db, direction="买", offset="开", price=7800, volume=1)
+        close_rec = _make_record(db, direction="卖", offset="平", price=7900, volume=1,
+                                 close_pnl=1000, commission=10)
+        pair = create_manual_pair(db, [open_rec.id], [close_rec.id])
+        reflection = TradeReflection(
+            trade_pair_id=pair.id,
+            trade_date="2026-07-05",
+            entry_reason="突破买入",
+            entry_timing="good",
+            discipline_score=4,
+            free_reflection="执行较好",
+        )
+        db.add(reflection)
+        db.commit()
+
+        prompt = build_reflection_prompt(db, pair, reflection)
+
+        assert "7800" in prompt
+        assert "7900" in prompt
+        assert "突破买入" in prompt
+        assert "执行较好" in prompt
+        assert "JSON" in prompt
+
+    def test_parse_ai_response_with_json(self):
+        """正常 JSON 响应解析"""
+        response = json.dumps({
+            "overall_evaluation": "综合评价",
+            "strengths": ["优点1"],
+            "weaknesses": ["不足1"],
+            "experience_suggestion": "建议",
+            "suggestion_type": "lesson",
+        })
+        parsed = _parse_ai_response(response)
+        assert parsed["overall_evaluation"] == "综合评价"
+        assert parsed["strengths"] == ["优点1"]
+
+    def test_parse_ai_response_with_markdown(self):
+        """Markdown 代码块中的 JSON 解析"""
+        response = '```json\n{"overall_evaluation": "评价", "strengths": [], "weaknesses": [], "experience_suggestion": "", "suggestion_type": "tip"}\n```'
+        parsed = _parse_ai_response(response)
+        assert parsed["suggestion_type"] == "tip"
+
+    def test_parse_ai_response_fallback(self):
+        """无法解析时返回原始内容"""
+        response = "这是 AI 的纯文本回复"
+        parsed = _parse_ai_response(response)
+        assert parsed["overall_evaluation"] == response
+
+
+class TestExperienceSave:
+    """经验保存测试"""
+
+    def test_save_suggestion_as_experience(self, db):
+        """将 AI 分析建议保存为经验"""
+        open_rec = _make_record(db, direction="买", offset="开", price=7800, volume=1)
+        close_rec = _make_record(db, direction="卖", offset="平", price=7900, volume=1,
+                                 close_pnl=1000, commission=10)
+        pair = create_manual_pair(db, [open_rec.id], [close_rec.id])
+        analysis = TradeAIAnalysis(
+            trade_pair_id=pair.id,
+            version=1,
+            overall_evaluation="评价",
+            experience_suggestion="重要经验",
+            suggestion_type="lesson",
+        )
+        db.add(analysis)
+        db.commit()
+
+        result = save_suggestion_as_experience(db, analysis.id, tags=["突破交易"])
+
+        assert result["success"] is True
+        exp = db.query(TradeExperience).first()
+        assert exp is not None
+        assert exp.exp_type == "lesson"
+        assert "重要经验" in exp.content
+        assert analysis.saved_to_experience is True
+        assert analysis.experience_id == exp.id