|
|
"""
|
|
|
交易反思系统单元测试与集成测试
|
|
|
"""
|
|
|
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
|
