|
|
"""
|
|
|
交易反思 AI 分析服务 - 基于反思内容重新分析交易并提炼经验
|
|
|
"""
|
|
|
import json
|
|
|
import logging
|
|
|
import re
|
|
|
from typing import Dict, List, Optional
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.analysis_models import TradePair, TradeReflection, TradeAIAnalysis, TradeExperience
|
|
|
from app.services.ai_analysis import AIFuturesAnalyzer
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
REFLECTION_ANALYSIS_PROMPT = """你是一位资深的交易心理与策略复盘教练。请基于以下交易数据和交易者的反思内容，进行深度分析。
|
|
|
|
|
|
=== 交易基础信息 ===
|
|
|
品种：{variety}
|
|
|
方向：{direction_text}
|
|
|
开仓日期时间：{open_date} {open_time}
|
|
|
平仓日期时间：{close_date} {close_time}
|
|
|
开仓均价：{open_price}
|
|
|
平仓均价：{close_price}
|
|
|
交易手数：{volume}
|
|
|
平仓盈亏：{close_pnl}
|
|
|
手续费：{commission}
|
|
|
净盈亏：{net_pnl}
|
|
|
|
|
|
=== 交易者反思内容 ===
|
|
|
入场理由：{entry_reason}
|
|
|
入场时机评价：{entry_timing}
|
|
|
仓位管理反思：{position_management}
|
|
|
出场理由：{exit_reason}
|
|
|
出场时机评价：{exit_timing}
|
|
|
纪律评分（1-5）：{discipline_score}
|
|
|
自由反思：{free_reflection}
|
|
|
|
|
|
=== 当日整体反思 ===
|
|
|
情绪状态：{emotion_state}
|
|
|
市场判断：{market_judgment}
|
|
|
当日总结：{daily_summary}
|
|
|
改进方向：{daily_improvements}
|
|
|
|
|
|
=== 任务要求 ===
|
|
|
请严格按以下 JSON 格式输出分析结果（不要输出任何其他内容）：
|
|
|
|
|
|
{{
|
|
|
"overall_evaluation": "综合评价：这笔交易的执行质量、决策逻辑、与市场的匹配度等",
|
|
|
"strengths": ["优点1", "优点2", "优点3"],
|
|
|
"weaknesses": ["不足1", "不足2", "不足3"],
|
|
|
"experience_suggestion": "提炼出的核心经验教训或改进建议，语言简洁 actionable",
|
|
|
"suggestion_type": "lesson",
|
|
|
"actionable_plan": ["下次可以做的具体改进1", "下次可以做的具体改进2"]
|
|
|
}}
|
|
|
|
|
|
注意：
|
|
|
1. suggestion_type 只能是 lesson（经验）、tip（技巧）、warning（警告）三者之一
|
|
|
2. 必须结合反思内容进行分析，不要脱离交易者自己的总结
|
|
|
3. 如果交易亏损，重点分析亏损原因和避免方案
|
|
|
4. 如果交易盈利，重点分析盈利是否来自计划还是运气
|
|
|
"""
|
|
|
|
|
|
|
|
|
def _timing_text(timing: Optional[str]) -> str:
|
|
|
"""时机评价转中文"""
|
|
|
mapping = {"good": "良好", "fair": "一般", "poor": "较差"}
|
|
|
return mapping.get(timing, timing or "未评价")
|
|
|
|
|
|
|
|
|
def _direction_text(direction: Optional[str]) -> str:
|
|
|
"""方向转中文"""
|
|
|
mapping = {"long": "多头", "short": "空头"}
|
|
|
return mapping.get(direction, direction or "未知")
|
|
|
|
|
|
|
|
|
def _get_records_text(db: Session, pair: TradePair) -> str:
|
|
|
"""获取配对关联的原始记录信息"""
|
|
|
from app.analysis_models import TradeRecord
|
|
|
record_ids = (pair.open_record_ids or []) + (pair.close_record_ids or [])
|
|
|
if not record_ids:
|
|
|
return ""
|
|
|
records = db.query(TradeRecord).filter(TradeRecord.id.in_(record_ids)).all()
|
|
|
lines = []
|
|
|
for r in sorted(records, key=lambda x: f"{x.trade_date} {x.trade_time or ''}"):
|
|
|
lines.append(f"{r.trade_date} {r.trade_time or ''} {r.symbol} {r.direction}{r.offset} {r.price}×{r.volume}")
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
|
|
|
def build_reflection_prompt(db: Session, pair: TradePair, reflection: TradeReflection,
|
|
|
daily_reflection: Optional[Dict] = None) -> str:
|
|
|
"""构建反思增强分析提示词"""
|
|
|
daily = daily_reflection or {}
|
|
|
|
|
|
# 获取第一条开仓和第一条平仓记录的时间
|
|
|
from app.analysis_models import TradeRecord
|
|
|
open_time = ""
|
|
|
close_time = ""
|
|
|
open_record_ids = pair.open_record_ids or []
|
|
|
close_record_ids = pair.close_record_ids or []
|
|
|
if open_record_ids:
|
|
|
open_rec = db.query(TradeRecord).filter(TradeRecord.id == open_record_ids[0]).first()
|
|
|
if open_rec:
|
|
|
open_time = open_rec.trade_time or ""
|
|
|
if close_record_ids:
|
|
|
close_rec = db.query(TradeRecord).filter(TradeRecord.id == close_record_ids[0]).first()
|
|
|
if close_rec:
|
|
|
close_time = close_rec.trade_time or ""
|
|
|
|
|
|
return REFLECTION_ANALYSIS_PROMPT.format(
|
|
|
variety=pair.variety,
|
|
|
direction_text=_direction_text(pair.direction),
|
|
|
open_date=pair.open_date or "",
|
|
|
open_time=open_time,
|
|
|
close_date=pair.close_date or "",
|
|
|
close_time=close_time,
|
|
|
open_price=pair.open_price,
|
|
|
close_price=pair.close_price,
|
|
|
volume=pair.total_volume,
|
|
|
close_pnl=pair.close_pnl,
|
|
|
commission=pair.total_commission,
|
|
|
net_pnl=pair.net_pnl,
|
|
|
entry_reason=reflection.entry_reason or "未填写",
|
|
|
entry_timing=_timing_text(reflection.entry_timing),
|
|
|
position_management=reflection.position_management or "未填写",
|
|
|
exit_reason=reflection.exit_reason or "未填写",
|
|
|
exit_timing=_timing_text(reflection.exit_timing),
|
|
|
discipline_score=reflection.discipline_score or "未评分",
|
|
|
free_reflection=reflection.free_reflection or "未填写",
|
|
|
emotion_state=daily.get("emotion_state", "未填写"),
|
|
|
market_judgment=daily.get("market_judgment", "未填写"),
|
|
|
daily_summary=daily.get("summary", "未填写"),
|
|
|
daily_improvements=daily.get("improvements", "未填写"),
|
|
|
)
|
|
|
|
|
|
|
|
|
def _parse_ai_response(response: str) -> Dict:
|
|
|
"""解析 AI 返回的 JSON"""
|
|
|
try:
|
|
|
# 尝试直接解析
|
|
|
data = json.loads(response)
|
|
|
return data
|
|
|
except json.JSONDecodeError:
|
|
|
pass
|
|
|
|
|
|
# 尝试从 Markdown 代码块中提取
|
|
|
code_block_pattern = re.compile(r'```(?:json)?\s*([\s\S]*?)```', re.MULTILINE)
|
|
|
matches = code_block_pattern.findall(response)
|
|
|
for match in matches:
|
|
|
try:
|
|
|
return json.loads(match.strip())
|
|
|
except json.JSONDecodeError:
|
|
|
continue
|
|
|
|
|
|
# 尝试从文本中提取第一个 JSON 对象
|
|
|
json_pattern = re.compile(r'\{[\s\S]*\}')
|
|
|
matches = json_pattern.findall(response)
|
|
|
for match in matches:
|
|
|
try:
|
|
|
return json.loads(match)
|
|
|
except json.JSONDecodeError:
|
|
|
continue
|
|
|
|
|
|
# 解析失败，返回原始内容作为 overall_evaluation
|
|
|
return {
|
|
|
"overall_evaluation": response,
|
|
|
"strengths": [],
|
|
|
"weaknesses": [],
|
|
|
"experience_suggestion": "",
|
|
|
"suggestion_type": "lesson",
|
|
|
"actionable_plan": [],
|
|
|
}
|
|
|
|
|
|
|
|
|
def _get_next_version(db: Session, trade_pair_id: int) -> int:
|
|
|
"""获取下一个版本号"""
|
|
|
latest = db.query(TradeAIAnalysis).filter(
|
|
|
TradeAIAnalysis.trade_pair_id == trade_pair_id
|
|
|
).order_by(TradeAIAnalysis.version.desc()).first()
|
|
|
return (latest.version + 1) if latest else 1
|
|
|
|
|
|
|
|
|
def analyze_with_reflection(db: Session, trade_pair_id: int) -> Dict:
|
|
|
"""
|
|
|
基于反思内容重新分析交易
|
|
|
返回：{success, data/version, message}
|
|
|
"""
|
|
|
pair = db.query(TradePair).filter(TradePair.id == trade_pair_id).first()
|
|
|
if not pair:
|
|
|
return {"success": False, "message": "交易配对不存在"}
|
|
|
|
|
|
reflection = db.query(TradeReflection).filter(
|
|
|
TradeReflection.trade_pair_id == trade_pair_id
|
|
|
).first()
|
|
|
if not reflection:
|
|
|
return {"success": False, "message": "该交易尚未填写反思，无法进行分析"}
|
|
|
|
|
|
# 获取当日反思
|
|
|
daily_reflection = None
|
|
|
from app.analysis_models import DailyReflection
|
|
|
dr = db.query(DailyReflection).filter(
|
|
|
DailyReflection.reflection_date == pair.close_date
|
|
|
).first()
|
|
|
if dr:
|
|
|
daily_reflection = {
|
|
|
"emotion_state": dr.emotion_state,
|
|
|
"market_judgment": dr.market_judgment,
|
|
|
"summary": dr.summary,
|
|
|
"improvements": dr.improvements,
|
|
|
}
|
|
|
|
|
|
# 构建提示词
|
|
|
prompt = build_reflection_prompt(db, pair, reflection, daily_reflection)
|
|
|
|
|
|
# 调用 AI
|
|
|
analyzer = AIFuturesAnalyzer(db)
|
|
|
model = analyzer.get_active_model()
|
|
|
if not model:
|
|
|
return {"success": False, "message": "未配置AI模型或模型未激活"}
|
|
|
|
|
|
response = analyzer.call_ai_model(prompt, model)
|
|
|
if not response:
|
|
|
return {"success": False, "message": "AI模型返回空响应，请稍后重试"}
|
|
|
|
|
|
# 解析结果
|
|
|
parsed = _parse_ai_response(response)
|
|
|
|
|
|
# 计算版本号
|
|
|
version = _get_next_version(db, trade_pair_id)
|
|
|
|
|
|
# 保存分析结果
|
|
|
analysis = TradeAIAnalysis(
|
|
|
trade_pair_id=trade_pair_id,
|
|
|
version=version,
|
|
|
overall_evaluation=parsed.get("overall_evaluation", ""),
|
|
|
strengths=parsed.get("strengths", []),
|
|
|
weaknesses=parsed.get("weaknesses", []),
|
|
|
experience_suggestion=parsed.get("experience_suggestion", ""),
|
|
|
suggestion_type=parsed.get("suggestion_type", "lesson"),
|
|
|
raw_response=response,
|
|
|
prompt_snapshot=prompt,
|
|
|
)
|
|
|
db.add(analysis)
|
|
|
|
|
|
# 更新反思的 AI 分析状态
|
|
|
reflection.ai_analyzed = True
|
|
|
reflection.ai_version = version
|
|
|
|
|
|
db.commit()
|
|
|
|
|
|
return {
|
|
|
"success": True,
|
|
|
"data": {
|
|
|
"version": version,
|
|
|
"overall_evaluation": analysis.overall_evaluation,
|
|
|
"strengths": analysis.strengths,
|
|
|
"weaknesses": analysis.weaknesses,
|
|
|
"experience_suggestion": analysis.experience_suggestion,
|
|
|
"suggestion_type": analysis.suggestion_type,
|
|
|
"actionable_plan": parsed.get("actionable_plan", []),
|
|
|
},
|
|
|
}
|
|
|
|
|
|
|
|
|
def get_analysis_history(db: Session, trade_pair_id: int) -> List[Dict]:
|
|
|
"""获取交易的所有 AI 分析版本"""
|
|
|
analyses = db.query(TradeAIAnalysis).filter(
|
|
|
TradeAIAnalysis.trade_pair_id == trade_pair_id
|
|
|
).order_by(TradeAIAnalysis.version.desc()).all()
|
|
|
|
|
|
return [{
|
|
|
"id": a.id,
|
|
|
"version": a.version,
|
|
|
"overall_evaluation": a.overall_evaluation,
|
|
|
"strengths": a.strengths,
|
|
|
"weaknesses": a.weaknesses,
|
|
|
"experience_suggestion": a.experience_suggestion,
|
|
|
"suggestion_type": a.suggestion_type,
|
|
|
"saved_to_experience": a.saved_to_experience,
|
|
|
"experience_id": a.experience_id,
|
|
|
"created_at": a.created_at.strftime('%Y-%m-%d %H:%M:%S') if a.created_at else None,
|
|
|
} for a in analyses]
|
|
|
|
|
|
|
|
|
def get_latest_analysis(db: Session, trade_pair_id: int) -> Optional[Dict]:
|
|
|
"""获取最新 AI 分析结果"""
|
|
|
analysis = db.query(TradeAIAnalysis).filter(
|
|
|
TradeAIAnalysis.trade_pair_id == trade_pair_id
|
|
|
).order_by(TradeAIAnalysis.version.desc()).first()
|
|
|
|
|
|
if not analysis:
|
|
|
return None
|
|
|
|
|
|
return {
|
|
|
"id": analysis.id,
|
|
|
"version": analysis.version,
|
|
|
"overall_evaluation": analysis.overall_evaluation,
|
|
|
"strengths": analysis.strengths,
|
|
|
"weaknesses": analysis.weaknesses,
|
|
|
"experience_suggestion": analysis.experience_suggestion,
|
|
|
"suggestion_type": analysis.suggestion_type,
|
|
|
"saved_to_experience": analysis.saved_to_experience,
|
|
|
"experience_id": analysis.experience_id,
|
|
|
"created_at": analysis.created_at.strftime('%Y-%m-%d %H:%M:%S') if analysis.created_at else None,
|
|
|
}
|
|
|
|
|
|
|
|
|
def save_suggestion_as_experience(db: Session, analysis_id: int, title: Optional[str] = None,
|
|
|
content: Optional[str] = None, tags: Optional[List[str]] = None) -> Dict:
|
|
|
"""将 AI 分析建议保存为经验"""
|
|
|
analysis = db.query(TradeAIAnalysis).filter(TradeAIAnalysis.id == analysis_id).first()
|
|
|
if not analysis:
|
|
|
return {"success": False, "message": "分析记录不存在"}
|
|
|
|
|
|
pair = db.query(TradePair).filter(TradePair.id == analysis.trade_pair_id).first()
|
|
|
|
|
|
exp_title = title or analysis.experience_suggestion[:50] if analysis.experience_suggestion else "AI提炼经验"
|
|
|
exp_content = content or analysis.experience_suggestion or ""
|
|
|
|
|
|
exp = TradeExperience(
|
|
|
title=exp_title,
|
|
|
content=exp_content,
|
|
|
exp_type=analysis.suggestion_type or "lesson",
|
|
|
source_pair_id=analysis.trade_pair_id,
|
|
|
source_date=pair.close_date if pair else None,
|
|
|
tags=tags or [],
|
|
|
)
|
|
|
db.add(exp)
|
|
|
db.flush()
|
|
|
|
|
|
analysis.saved_to_experience = True
|
|
|
analysis.experience_id = exp.id
|
|
|
db.commit()
|
|
|
|
|
|
return {"success": True, "data": {"experience_id": exp.id}}
|
