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
refactor3.0
Lxy 2 weeks ago
parent 3d7ddd5892
commit a2ef70f607

@ -2,35 +2,54 @@
交易配对引擎 - 将开仓和平仓配对为完整交易持久化到 TradePair 交易配对引擎 - 将开仓和平仓配对为完整交易持久化到 TradePair
""" """
import logging import logging
from datetime import datetime
from typing import Optional from typing import Optional
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from collections import defaultdict 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__) logger = logging.getLogger(__name__)
def auto_pair_trades(db: Session, trade_date: str) -> dict: def auto_pair_trades(db: Session, trade_date: str) -> dict:
""" """
自动配对指定日期的交易记录 自动配对指定日期及之前未平仓的交易记录
按品种+方向分组FIFO 规则配对 按品种+方向分组FIFO 规则配对
返回{created: 新建配对数, updated: 更新配对数, unpaired: 未配对记录数} 返回{created: 新建配对数, updated: 更新配对数, unpaired: 未配对记录数}
""" """
records = db.query(TradeRecord).filter( # 查询指定日期的记录
current_records = db.query(TradeRecord).filter(
TradeRecord.trade_date == trade_date TradeRecord.trade_date == trade_date
).order_by( ).order_by(
TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time
).all() ).all()
if not records: if not current_records:
return {"created": 0, "updated": 0, "unpaired": 0} 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) by_variety = defaultdict(list)
for r in records: for r in all_records:
by_variety[r.variety].append(r) by_variety[r.variety].append(r)
result = {"created": 0, "updated": 0, "unpaired": 0} 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() all_paired_close_ids = set()
existing_pairs = db.query(TradePair).filter( existing_pairs = db.query(TradePair).filter(
TradePair.variety == variety, TradePair.variety == variety,
TradePair.open_date >= records[0].trade_date if records else None
).all() ).all()
for p in existing_pairs: for p in existing_pairs:
all_paired_open_ids.update(p.open_record_ids) all_paired_open_ids.update(p.open_record_ids or [])
all_paired_close_ids.update(p.close_record_ids) all_paired_close_ids.update(p.close_record_ids or [])
unpaired_count = 0 unpaired_count = 0
for r in records: 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: def _pair_direction(db: Session, variety: str, opens: list, closes: list, direction: str) -> int:
"""单方向配对FIFO返回新建配对数""" """单方向配对FIFO支持多开合并返回新建配对数
每个平仓记录会尽可能与前面的开仓记录配对形成一个完整配对
如果多个开仓合并才能匹配一个平仓则这些开仓记录会归入同一个配对
"""
created = 0 created = 0
open_queue = list(opens) # FIFO 队列 # 开仓队列,每个元素是 (record, remaining_volume)
close_queue = list(closes) 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: while open_queue and close_queue:
open_rec = open_queue[0] close_rec, close_remaining = close_queue[0]
close_rec = close_queue[0]
# 从开仓队列中取足够手数来匹配这个平仓
open_volume = open_rec.volume or 0 needed = close_remaining
close_volume = close_rec.volume or 0 used_opens = [] # (record, take_volume)
while needed > 0 and open_queue:
# 计算配对手数 open_rec, open_remaining = open_queue[0]
pair_volume = min(open_volume, close_volume) take = min(open_remaining, needed)
used_opens.append((open_rec, take))
# 计算平均价格 needed -= take
open_price = open_rec.price if take >= open_remaining:
close_price = close_rec.price open_queue.pop(0)
else:
# 计算盈亏 open_queue[0] = (open_rec, open_remaining - take)
close_pnl = close_rec.close_pnl or 0
total_commission = (open_rec.commission or 0) + (close_rec.commission or 0) 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 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( pair = TradePair(
variety=variety, variety=variety,
direction=direction, direction=direction,
open_record_ids=[open_rec.id], open_record_ids=[r.id for r, _ in used_opens],
close_record_ids=[close_rec.id], close_record_ids=[close_rec.id],
open_price=open_price, open_price=round(open_price, 2),
close_price=close_price, close_price=round(close_price, 2),
total_volume=pair_volume, total_volume=matched_volume,
close_pnl=close_pnl, close_pnl=round(close_pnl, 2),
total_commission=total_commission, total_commission=round(total_commission, 2),
net_pnl=net_pnl, net_pnl=round(net_pnl, 2),
open_date=open_rec.trade_date, open_date=open_date,
close_date=close_rec.trade_date, close_date=close_date,
) )
db.add(pair) db.add(pair)
created += 1 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() db.flush()
return created return created
@ -200,8 +236,8 @@ def get_daily_trades_with_pairs(db: Session, trade_date: str) -> dict:
# 未配对记录 # 未配对记录
paired_record_ids = set() paired_record_ids = set()
for p in pairs: for p in pairs:
paired_record_ids.update(p.open_record_ids) paired_record_ids.update(p.open_record_ids or [])
paired_record_ids.update(p.close_record_ids) paired_record_ids.update(p.close_record_ids or [])
unpaired = [r for r in records if r.id not in paired_record_ids] unpaired = [r for r in records if r.id not in paired_record_ids]
unpaired_data = [{ unpaired_data = [{

@ -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
Loading…
Cancel
Save