You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
buffer_platform/app/api/trade_review.py

591 lines
22 KiB

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

"""
交易复盘接口 - 提供交易记录导入、查询、汇总、统计等功能
"""
import logging
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.analysis_db import get_analysis_db
from app.analysis_models import TradeRecord, TradeImportBatch
from app.services.trade_parser import (
parse_settlement_file,
save_to_db,
calc_daily_summary,
calc_variety_summary,
calc_overall_statistics,
get_trade_pairs,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/trade-review", tags=["交易复盘"])
# ==================== 文件导入 ====================
@router.post("/import")
async def import_settlement(
file: UploadFile = File(...),
db: Session = Depends(get_analysis_db),
):
"""上传并解析期货结算单文件,导入交易记录(含逐条去重校验)"""
if not file.filename.endswith(('.xls', '.xlsx')):
raise HTTPException(status_code=400, detail="仅支持 .xls/.xlsx 格式的结算单文件")
try:
content = await file.read()
df_futures, df_options = parse_settlement_file(content, file.filename)
if df_futures.empty and df_options.empty:
return {"success": False, "message": "文件中未找到有效的交易记录(请确认是期货公司导出的结算单)"}
result = save_to_db(db, df_futures, df_options, file.filename)
total_skipped = result['futures_skipped'] + result['options_skipped']
total_new = result['futures_count'] + result['options_count']
total_parsed = result['futures_parsed'] + result['options_parsed']
if total_new == 0 and total_skipped > 0:
return {
"success": True,
"message": f"全部跳过:共解析 {total_parsed} 条记录,均已存在(去重规则:交易日+品种+时间+价格),未导入任何新数据",
"data": result,
}
msg_parts = [f"导入成功:期货 {result['futures_count']} 条,期权 {result['options_count']}"]
if total_skipped > 0:
msg_parts.append(f"跳过重复 {total_skipped}")
msg_parts.append(f"交易日期: {result['trade_dates']}")
return {
"success": True,
"message": "".join(msg_parts),
"data": result,
}
except HTTPException:
raise
except Exception as e:
logger.exception("导入结算单失败")
raise HTTPException(status_code=500, detail=f"导入失败: {str(e)}")
@router.post("/batch-import")
async def batch_import_settlement(
files: list[UploadFile] = File(...),
db: Session = Depends(get_analysis_db),
):
"""批量上传并解析期货结算单文件,导入交易记录(含逐条去重校验)"""
if not files:
raise HTTPException(status_code=400, detail="请选择至少一个结算单文件")
results = []
total_futures = 0
total_options = 0
total_skipped_all = 0
processed_count = 0
failed_count = 0
for file in files:
if not file.filename.endswith(('.xls', '.xlsx')):
results.append({
"filename": file.filename,
"success": False,
"message": "文件格式不支持,仅支持 .xls/.xlsx",
})
failed_count += 1
continue
try:
content = await file.read()
df_futures, df_options = parse_settlement_file(content, file.filename)
if df_futures.empty and df_options.empty:
results.append({
"filename": file.filename,
"success": False,
"message": "文件中未找到有效的交易记录",
})
failed_count += 1
continue
result = save_to_db(db, df_futures, df_options, file.filename)
file_new = result['futures_count'] + result['options_count']
file_skipped = result['futures_skipped'] + result['options_skipped']
total_futures += result['futures_count']
total_options += result['options_count']
total_skipped_all += file_skipped
processed_count += 1
msg_parts = []
if file_new > 0:
msg_parts.append(f"新导入 {file_new} 条(期货 {result['futures_count']},期权 {result['options_count']}")
if file_skipped > 0:
msg_parts.append(f"跳过重复 {file_skipped}")
if not msg_parts:
msg_parts.append("无有效记录")
results.append({
"filename": file.filename,
"success": True,
"message": "".join(msg_parts),
"new_count": file_new,
"skipped_count": file_skipped,
"trade_dates": result['trade_dates'],
"batch_id": result['batch_id'],
})
except Exception as e:
logger.exception(f"批量导入文件失败: {file.filename}")
results.append({
"filename": file.filename,
"success": False,
"message": f"导入失败: {str(e)}",
})
failed_count += 1
summary = (
f"批量导入完成:共 {len(files)} 个文件,"
f"处理 {processed_count} 个,失败 {failed_count} 个,"
f"合计新导入期货 {total_futures} 条、期权 {total_options} 条,跳过重复 {total_skipped_all}"
)
return {
"success": True,
"message": summary,
"data": {
"total_files": len(files),
"processed_count": processed_count,
"failed_count": failed_count,
"total_futures": total_futures,
"total_options": total_options,
"total_skipped": total_skipped_all,
"details": results,
},
}
# ==================== 交易记录查询 ====================
@router.get("/records")
def get_records(
start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
symbol: Optional[str] = Query(None, description="合约代码"),
variety: Optional[str] = Query(None, description="品种代码"),
trade_type: Optional[str] = Query(None, description="类型: 期货/期权"),
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
db: Session = Depends(get_analysis_db),
):
"""查询交易记录(支持筛选和分页)"""
query = db.query(TradeRecord)
if start_date:
query = query.filter(TradeRecord.trade_date >= start_date)
if end_date:
query = query.filter(TradeRecord.trade_date <= end_date)
if symbol:
query = query.filter(TradeRecord.symbol == symbol)
if variety:
query = query.filter(TradeRecord.variety == variety)
if trade_type:
query = query.filter(TradeRecord.trade_type == trade_type)
total = query.count()
records = query.order_by(
TradeRecord.trade_date.desc(),
TradeRecord.trade_time.desc(),
).offset((page - 1) * page_size).limit(page_size).all()
return {
"success": True,
"data": {
"total": total,
"page": page,
"page_size": page_size,
"records": [{
"id": r.id,
"trade_type": r.trade_type,
"symbol": r.symbol,
"variety": r.variety,
"symbol_name": r.symbol_name,
"direction": r.direction,
"offset": r.offset,
"price": r.price,
"volume": r.volume,
"amount": r.amount,
"close_pnl": r.close_pnl,
"commission": r.commission,
"trade_date": r.trade_date,
"trade_time": r.trade_time,
"import_batch": r.import_batch,
"source_file": r.source_file,
} for r in records],
},
}
# ==================== 批次管理 ====================
@router.get("/batches")
def get_batches(
db: Session = Depends(get_analysis_db),
):
"""查询导入批次列表"""
batches = db.query(TradeImportBatch).order_by(
TradeImportBatch.created_at.desc()
).all()
return {
"success": True,
"data": [{
"id": b.id,
"batch_id": b.batch_id,
"source_file": b.source_file,
"futures_count": b.futures_count,
"options_count": b.options_count,
"trade_dates": b.trade_dates,
"created_at": b.created_at.strftime('%Y-%m-%d %H:%M:%S') if b.created_at else '',
} for b in batches],
}
@router.delete("/records/{batch_id}")
def delete_batch_records(
batch_id: str,
db: Session = Depends(get_analysis_db),
):
"""按批次删除交易记录"""
count = db.query(TradeRecord).filter(TradeRecord.import_batch == batch_id).count()
if count == 0:
raise HTTPException(status_code=404, detail="未找到该批次的交易记录")
db.query(TradeRecord).filter(TradeRecord.import_batch == batch_id).delete()
db.query(TradeImportBatch).filter(TradeImportBatch.batch_id == batch_id).delete()
db.commit()
return {"success": True, "message": f"已删除 {count} 条交易记录"}
# ==================== 汇总统计 ====================
@router.get("/latest-trade-date")
def get_latest_trade_date(
db: Session = Depends(get_analysis_db),
):
"""获取最后一个有交易记录的交易日"""
from datetime import date
latest = db.query(TradeRecord.trade_date).filter(
TradeRecord.trade_date.isnot(None),
TradeRecord.trade_date != '',
).distinct().order_by(TradeRecord.trade_date.desc()).first()
if latest:
return {"success": True, "data": {"trade_date": latest[0]}}
else:
return {"success": True, "data": {"trade_date": date.today().strftime('%Y-%m-%d')}}
@router.delete("/records-by-date/{trade_date}")
def delete_records_by_date(
trade_date: str,
db: Session = Depends(get_analysis_db),
):
"""删除指定交易日的所有交易记录"""
count = db.query(TradeRecord).filter(TradeRecord.trade_date == trade_date).count()
if count == 0:
raise HTTPException(status_code=404, detail=f"未找到 {trade_date} 的交易记录")
db.query(TradeRecord).filter(TradeRecord.trade_date == trade_date).delete()
db.commit()
return {"success": True, "message": f"已删除 {trade_date}{count} 条交易记录"}
@router.get("/daily-summary")
def get_daily_summary(
start_date: Optional[str] = Query(None),
end_date: Optional[str] = Query(None),
db: Session = Depends(get_analysis_db),
):
"""获取每日交易盈亏汇总"""
data = calc_daily_summary(db, start_date, end_date)
return {"success": True, "data": data}
@router.get("/variety-summary")
def get_variety_summary(
start_date: Optional[str] = Query(None),
end_date: Optional[str] = Query(None),
db: Session = Depends(get_analysis_db),
):
"""获取品种交易盈亏汇总"""
data = calc_variety_summary(db, start_date, end_date)
return {"success": True, "data": data}
@router.get("/statistics")
def get_statistics(
start_date: Optional[str] = Query(None),
end_date: Optional[str] = Query(None),
db: Session = Depends(get_analysis_db),
):
"""获取整体交易统计"""
data = calc_overall_statistics(db, start_date, end_date)
return {"success": True, "data": data}
@router.get("/trade-pairs")
def get_trade_pairs_api(
start_date: Optional[str] = Query(None),
end_date: Optional[str] = Query(None),
symbol: Optional[str] = Query(None),
db: Session = Depends(get_analysis_db),
):
"""获取开平仓配对后的逐笔交易"""
pairs = get_trade_pairs(db, start_date, end_date)
if symbol:
pairs = [p for p in pairs if p["symbol"] == symbol]
return {"success": True, "data": pairs}
# ==================== K线 + 交易标记 ====================
@router.get("/kline-with-trades/{symbol}")
def get_kline_with_trades(
symbol: str,
period: str = Query("daily", description="K线周期: daily/60min/15min/5min"),
start_date: Optional[str] = Query(None),
end_date: Optional[str] = Query(None),
db: Session = Depends(get_analysis_db),
):
"""获取K线数据 + 该品种的交易标记买卖点。symbol 可以是完整合约代码(AG2608)或品种代码(AG)"""
import json
import re
from app.database import SessionLocal as MainSessionLocal
from app.models import MarketData
logger.info(f"[K线查询] 请求参数: symbol={symbol}, period={period}, start_date={start_date}, end_date={end_date}")
# 判断是品种代码还是完整合约代码
variety_code = symbol.upper()
if re.match(r'^[A-Za-z]+\d', symbol):
# 完整合约代码如 AG2608
variety_code = re.match(r'^([A-Za-z]+)', symbol).group(1).upper()
kline_symbol = symbol
else:
# 品种代码如 AG从配置中查找当前合约
kline_symbol = _resolve_symbol_from_config(variety_code)
logger.info(f"[K线查询] 解析结果: variety_code={variety_code}, kline_symbol={kline_symbol}")
main_db = MainSessionLocal()
try:
# 尝试精确匹配先尝试原始symbol再尝试小写
market_data = main_db.query(MarketData).filter(
MarketData.symbol == kline_symbol,
MarketData.period == period,
).first()
logger.info(f"[K线查询] 精确匹配查询: symbol={kline_symbol}, period={period}, found={market_data is not None}")
if not market_data or not market_data.candles_json:
# 尝试小写匹配
market_data = main_db.query(MarketData).filter(
MarketData.symbol == kline_symbol.lower(),
MarketData.period == period,
).first()
logger.info(f"[K线查询] 小写匹配查询: symbol={kline_symbol.lower()}, period={period}, found={market_data is not None}")
if not market_data or not market_data.candles_json:
# 尝试模糊匹配:查找以该品种代码开头的合约(不区分大小写)
market_data = main_db.query(MarketData).filter(
MarketData.symbol.like(f'{variety_code}%'),
MarketData.period == period,
).first()
logger.info(f"[K线查询] 模糊匹配查询: like '{variety_code}%', period={period}, found={market_data is not None}")
if market_data:
logger.info(f"[K线查询] 模糊匹配找到: symbol={market_data.symbol}")
if not market_data or not market_data.candles_json:
# 尝试小写模糊匹配
market_data = main_db.query(MarketData).filter(
MarketData.symbol.like(f'{variety_code.lower()}%'),
MarketData.period == period,
).first()
logger.info(f"[K线查询] 小写模糊匹配查询: like '{variety_code.lower()}%', period={period}, found={market_data is not None}")
if market_data:
logger.info(f"[K线查询] 小写模糊匹配找到: symbol={market_data.symbol}")
if not market_data or not market_data.candles_json:
logger.warning(f"[K线查询] 未找到K线数据: {kline_symbol} {period}")
return {"success": False, "message": f"未找到 {kline_symbol}{period} K线数据"}
candles = json.loads(market_data.candles_json)
# 将字典格式K线转换为前端期望的数组格式 [date, open, close, low, high, volume]
if candles and isinstance(candles[0], dict):
# 根据周期决定是否保留时间部分
is_daily = period == 'daily'
candles = [
[
c.get("time", "")[:10] if is_daily else c.get("time", ""), # 日线只保留日期,分钟线保留完整时间
c.get("open", 0),
c.get("close", 0),
c.get("low", 0),
c.get("high", 0),
c.get("volume", 0),
]
for c in candles
]
kline_symbol = market_data.symbol
logger.info(f"[K线查询] 成功获取K线数据: symbol={kline_symbol}, candles_count={len(candles)}")
finally:
main_db.close()
# 获取该品种的交易记录
logger.info(f"[K线查询] 开始查询交易记录: variety={variety_code}, start_date={start_date}, end_date={end_date}")
query = db.query(TradeRecord).filter(TradeRecord.variety == variety_code)
if start_date:
query = query.filter(TradeRecord.trade_date >= start_date)
if end_date:
query = query.filter(TradeRecord.trade_date <= end_date)
trades = query.order_by(TradeRecord.trade_date, TradeRecord.trade_time).all()
logger.info(f"[K线查询] 交易记录查询完成: 找到 {len(trades)} 条记录")
if len(trades) > 0:
logger.info(f"[K线查询] 交易记录示例: {[{'symbol': t.symbol, 'variety': t.variety, 'date': t.trade_date, 'direction': t.direction} for t in trades[:3]]}")
trade_markers = []
for t in trades:
trade_markers.append({
"date": t.trade_date,
"time": t.trade_time,
"symbol": t.symbol,
"direction": t.direction,
"offset": t.offset,
"price": t.price,
"volume": t.volume,
"close_pnl": t.close_pnl,
"commission": t.commission,
})
logger.info(f"[K线查询] 返回数据: symbol={kline_symbol}, period={period}, candles={len(candles)}, markers={len(trade_markers)}")
return {
"success": True,
"data": {
"symbol": kline_symbol,
"period": period,
"candles": candles,
"trade_markers": trade_markers,
},
}
def _resolve_symbol_from_config(variety_code: str) -> str:
"""从品种配置中查找品种代码对应的当前合约"""
from app.config_store import get_config_store
config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
for name, contract in config.get("futures", {}).items():
if contract.upper().startswith(variety_code.upper()):
return contract
return variety_code
# ==================== AI 逐笔交易分析 ====================
class AnalyzeTradeRequest(BaseModel):
symbol: str
open_date: str
open_time: Optional[str] = None
close_date: Optional[str] = None
close_time: Optional[str] = None
direction: str # 多/空
open_price: Optional[float] = None
close_price: Optional[float] = None
@router.post("/analyze-trade")
async def analyze_trade(
req: AnalyzeTradeRequest,
db: Session = Depends(get_analysis_db),
):
"""AI 分析单笔交易结合多周期K线数据"""
# 收集多周期K线数据
from app.database import SessionLocal as MainSessionLocal
from app.models import MarketData
import json
periods = ["daily", "60min", "15min", "5min"]
kline_context = {}
main_db = MainSessionLocal()
try:
for period in periods:
market_data = main_db.query(MarketData).filter(
MarketData.symbol == req.symbol,
MarketData.period == period,
).first()
if market_data and market_data.candles_json:
candles = json.loads(market_data.candles_json)
# 将字典格式K线转换为数组格式 [date, open, close, low, high, volume]
if candles and isinstance(candles[0], dict):
candles = [
[c.get("time", "")[:10], c.get("open", 0), c.get("close", 0),
c.get("low", 0), c.get("high", 0), c.get("volume", 0)]
for c in candles
]
kline_context[period] = candles[-50:] if len(candles) > 50 else candles
finally:
main_db.close()
if not kline_context:
return {"success": False, "message": f"未找到 {req.symbol} 的K线数据无法分析"}
# 构建分析提示
prompt = f"""请分析以下期货交易交易的优缺点:
交易信息:
- 品种:{req.symbol}
- 方向:{req.direction}
- 开仓时间:{req.open_date} {req.open_time or ''}
- 开仓价格:{req.open_price or '未知'}
- 平仓时间:{req.close_date or '未知'} {req.close_time or ''}
- 平仓价格:{req.close_price or '未知'}
各周期K线数据格式[日期, 开盘, 收盘, 最低, 最高, 成交量]
"""
for period, candles in kline_context.items():
prompt += f"\n{period} 周期(最近{len(candles)}根K线\n"
for c in candles[-10:]:
prompt += f" {c[0]}: 开{c[1]}{c[2]}{c[3]}{c[4]}{c[5]}\n"
prompt += """
请从以下维度分析这笔交易:
1. 入场时机分析:入场时各周期的趋势如何,入场点是否合理
2. 出场时机分析:出场时的市场状态,是否过早/过晚出场
3. 持仓期间分析:持仓期间市场走势,是否有更好的操作机会
4. 综合评价:这笔交易的主要优点和不足之处
5. 改进建议:类似行情下如何优化操作
"""
try:
# 使用 AIFuturesAnalyzer 的模型配置和调用能力
from app.services.ai_analysis import AIFuturesAnalyzer
analyzer = AIFuturesAnalyzer(db)
model = analyzer.get_active_model()
if not model:
return {"success": False, "message": "未配置AI模型或模型未激活请先在AI配置页面设置"}
response = analyzer.call_ai_model(prompt, model)
if not response:
return {"success": False, "message": "AI模型返回空响应请稍后重试"}
return {"success": True, "data": {"analysis": response, "symbol": req.symbol}}
except Exception as e:
logger.exception("AI分析交易失败")
return {"success": False, "message": f"AI分析失败: {str(e)}"}