liaozhaorun/NewQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tqsdk实盘

Browse Source
This commit is contained in:
liaozhaorun
2025-07-10 15:07:31 +08:00
parent 4c243a4b47
commit 5de1a43b02
26 changed files with 12462 additions and 17535 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
src/analysis/result_analyzer.py
View File

@@ -2,8 +2,12 @@
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from typing import List, Dict, Any, Optional
from src.indicators.base_indicators import Indicator
# 导入纯函数 (注意相对导入路径的变化)
from .analysis_utils import (
calculate_metrics,
@@ -26,6 +30,7 @@ class ResultAnalyzer:
trade_history: List[Trade],
bars: List[Bar],
initial_capital: float,
indicator_list: List[Indicator] = [],
):
"""
Args:
@@ -37,8 +42,9 @@ class ResultAnalyzer:
self.portfolio_snapshots = portfolio_snapshots
self.trade_history = trade_history
self.initial_capital = initial_capital
self.bars = bars # 接收所有K线数据
self.bars = bars # 接收所有K线数据
self._metrics_cache: Optional[Dict[str, Any]] = None
self.indicator_list = indicator_list
print("\n--- 结果分析器初始化完成 ---")
@@ -63,14 +69,17 @@ class ResultAnalyzer:
print("\n--- 交易明细 ---")
for trade in self.trade_history:
# 调整输出格式，显示实现盈亏
pnl_display = f" | PnL: {trade.realized_pnl:.2f}" if trade.is_close_trade else ""
pnl_display = (
f" | Indicators:{trade.indicator_dict} | PnL: {trade.realized_pnl:.2f}"
if trade.is_close_trade
else ""
)
print(
f" {trade.fill_time} | {trade.direction:<10} | {trade.symbol} | Vol: {trade.volume} | Price: {trade.price:.2f} | Comm: {trade.commission:.2f}{pnl_display}"
)
else:
print("\n没有交易记录。")
metrics = self.calculate_all_metrics()
print("\n--- 回测绩效报告 ---")
@@ -82,7 +91,7 @@ class ResultAnalyzer:
print(f"{'夏普比率':<15}: {metrics['夏普比率']:.2f}")
print(f"{'卡玛比率':<15}: {metrics['卡玛比率']:.2f}")
print(f"{'总交易次数':<15}: {metrics['总交易次数']}")
print(f"{'总实现盈亏':<15}: {metrics['总实现盈亏']:.2f}") # 新增
print(f"{'总实现盈亏':<15}: {metrics['总实现盈亏']:.2f}") # 新增
print(f"{'交易成本':<15}: {metrics['交易成本']:.2f}")
# 新增交易相关详细指标，以适应更全面的交易分析需求
@@ -94,7 +103,6 @@ class ResultAnalyzer:
print(f"{'平均每次盈利':<15}: {metrics['平均每次盈利']:.2f}")
print(f"{'平均每次亏损':<15}: {metrics['平均每次亏损']:.2f}")
def plot_performance(self) -> None:
"""
绘制投资组合净值和回撤曲线，以及所有合约的收盘价曲线。
@@ -103,9 +111,200 @@ class ResultAnalyzer:
plot_equity_and_drawdown_chart(
self.portfolio_snapshots,
self.initial_capital,
title="Portfolio Equity and Drawdown Curve (All Contracts)" # 明确标题，表明是整体曲线
title="Portfolio Equity and Drawdown Curve (All Contracts)", # 明确标题，表明是整体曲线
)
# 绘制所有处理过的K线收盘价曲线
plot_close_price_chart(self.bars, title="Underlying Asset Close Price (Concatenated Bars)") # 明确标题
print("图表绘制完成。")
plot_close_price_chart(
self.bars, title="Underlying Asset Close Price (Concatenated Bars)"
) # 明确标题
print("图表绘制完成。")
def analyze_indicators(self):
"""
分析所有平仓交易的指标值与实现盈亏的关系，并绘制累积盈亏曲线图。
图表将展示指标值区间与对应累积盈亏的关系，帮助找出具有概率优势的指标区间。
同时会标记出最大和最小累积盈亏对应的指标值，并优化标注位置以避免重叠。
"""
close_trades = [trade for trade in self.trade_history if trade.is_close_trade]
if not close_trades:
print(
"没有平仓交易可供分析。请确保 trade_history 中有 is_close_trade 为 True 的交易。"
)
return
for indicator in self.indicator_list:
# 假设每个 indicator 对象都有一个 get_name() 方法
indicator_name = indicator.get_name()
# 收集该指标的所有值和对应的实现盈亏
indi_values = []
pnls = []
for trade in close_trades:
# 确保 trade.indicator_dict 中包含当前指标的值
# 并且这个值是可用的（非None或NaN）
if (
indicator_name in trade.indicator_dict
and trade.indicator_dict[indicator_name] is not None
):
# 检查是否为 NaN，如果使用 np.nan，则需要 isinstance(value, float) and np.isnan(value)
# 为了简化，这里假设非 None 即为有效数值
if not (
isinstance(trade.indicator_dict[indicator_name], float)
and np.isnan(trade.indicator_dict[indicator_name])
):
indi_values.append(trade.indicator_dict[indicator_name])
pnls.append(trade.realized_pnl)
if not indi_values:
print(f"指标 '{indicator_name}' 没有对应的有效平仓交易数据。跳过绘图。")
continue
# 将收集到的数据转换为 Pandas DataFrame 进行更便捷的处理
# DataFrame 的结构为：['indicator_value', 'realized_pnl']
df = pd.DataFrame({"indicator_value": indi_values, "realized_pnl": pnls})
# 确保数据框不为空
if df.empty:
print(f"指标 '{indicator_name}' 的数据框为空，跳过绘图。")
continue
# 按照指标值进行排序，这是计算累积和的关键步骤
df = df.sort_values(by="indicator_value").reset_index(drop=True)
# --- 绘制累积收益曲线 ---
plt.figure(figsize=(12, 7)) # 创建一个新的图表
# 获取指标值的范围，用于生成X轴的等距点
min_val = df["indicator_value"].min()
max_val = df["indicator_value"].max()
# 特殊处理：如果所有指标值都相同
if min_val == max_val:
total_pnl = df["realized_pnl"].sum()
print(
f"指标 '{indicator_name}' 的所有值都相同 ({min_val:.2f})，无法创建区间图，绘制一个点表示总收益。"
)
plt.plot(min_val, total_pnl, "ro", markersize=8) # 绘制一个点
plt.title(
f"{indicator_name} Value vs. Cumulative PnL (All values are {min_val:.2f})"
)
plt.xlabel(f"{indicator_name} Value")
plt.ylabel("Cumulative Realized PnL")
plt.grid(True)
plt.text(
min_val,
total_pnl,
f" Total PnL: {total_pnl:.2f}",
ha="center",
va="bottom",
)
plt.show()
continue
# 生成X轴上的100个等距点，这些点代表了指标值的不同阈值
x_points = np.linspace(min_val, max_val, 100)
# 计算Y轴的值：对于每个X轴点 xp，Y轴值是所有 'indicator_value' <= xp 的 'realized_pnl' 之和
y_cumulative_pnl = []
for xp in x_points:
# 筛选出指标值小于等于当前x_point的所有交易，并求和它们的 realized_pnl
cumulative_pnl = df[df["indicator_value"] <= xp]["realized_pnl"].sum()
y_cumulative_pnl.append(cumulative_pnl)
# 绘制累积盈亏曲线
plt.plot(
x_points,
y_cumulative_pnl,
marker="o",
linestyle="-",
markersize=3,
label=f"Cumulative PnL for {indicator_name}",
alpha=0.8,
)
# 标记累积盈亏的最大值点
optimal_index = np.argmax(y_cumulative_pnl)
optimal_indi_value = x_points[optimal_index]
max_cumulative_pnl = y_cumulative_pnl[optimal_index]
# 标记累积盈亏的最小值点
min_pnl_index = np.argmin(y_cumulative_pnl[:optimal_index]) if len(y_cumulative_pnl[:optimal_index]) > 0 else 0
min_indi_value_at_pnl = x_points[min_pnl_index]
min_cumulative_pnl = y_cumulative_pnl[min_pnl_index]
# 动态调整标注位置以避免重叠
offset_x = (max_val - min_val) * 0.05 # 水平偏移量
# 默认标注为右侧对齐，文本在点的左侧
max_ha = "right"
max_xytext_x = optimal_indi_value - offset_x
min_ha = "right"
min_xytext_x = min_indi_value_at_pnl - offset_x
# 如果最大值点在最小值点右侧，则最大值标注放左侧，最小值标注放右侧
# 这样可以避免标注文本重叠
if optimal_indi_value > min_indi_value_at_pnl:
max_ha = "left"
max_xytext_x = optimal_indi_value + offset_x
min_ha = "right"
min_xytext_x = min_indi_value_at_pnl - offset_x
else: # 如果最大值点在最小值点左侧或重合
max_ha = "right"
max_xytext_x = optimal_indi_value - offset_x
min_ha = "left"
min_xytext_x = min_indi_value_at_pnl + offset_x
# 绘制最大值垂直线和标注
plt.axvline(
optimal_indi_value,
color="red",
linestyle="--",
label=f"Max PnL Threshold: {optimal_indi_value:.2f}",
alpha=0.7,
)
plt.annotate(
f"Max Cum. PnL: {max_cumulative_pnl:.2f}",
xy=(optimal_indi_value, max_cumulative_pnl),
xytext=(max_xytext_x, max_cumulative_pnl),
arrowprops=dict(facecolor="red", shrink=0.05),
horizontalalignment=max_ha,
verticalalignment="bottom",
color="red",
)
# 绘制最小值垂直线和标注
plt.axvline(
min_indi_value_at_pnl,
color="blue",
linestyle=":",
label=f"Min PnL Threshold: {min_indi_value_at_pnl:.2f}",
alpha=0.7,
)
# 垂直偏移最小值标注，避免与曲线重叠
min_text_y_offset = (
-(max_cumulative_pnl - min_cumulative_pnl) * 0.1
if max_cumulative_pnl != min_cumulative_pnl
else -0.05
)
plt.annotate(
f"Min Cum. PnL: {min_cumulative_pnl:.2f}",
xy=(min_indi_value_at_pnl, min_cumulative_pnl),
xytext=(min_xytext_x, min_cumulative_pnl + min_text_y_offset),
arrowprops=dict(facecolor="blue", shrink=0.05),
horizontalalignment=min_ha,
verticalalignment="top",
color="blue",
)
plt.title(f"{indicator_name} Value vs. Cumulative Realized PnL")
plt.xlabel(f"{indicator_name} Value")
plt.ylabel("Cumulative Realized PnL")
plt.grid(True)
plt.legend()
plt.tight_layout() # 自动调整图表参数，使之更紧凑
plt.show()
print("\n所有指标的分析图表已生成。")