backtesting.py
from collections import defaultdict
from datetime import date, datetime
from copy import copy
from typing import cast
import traceback
import numpy as np
import polars as pl
import plotly.graph_objects as go # type: ignore
from plotly.subplots import make_subplots # type: ignore
from tqdm import tqdm
from vnpy.trader.constant import Direction, Offset, Interval, Status
from vnpy.trader.object import OrderData, TradeData, BarData
from vnpy.trader.utility import round_to, extract_vt_symbol
from ..logger import logger
from ..lab import AlphaLab
from .template import AlphaStrategy
class BacktestingEngine:
"""Alpha strategy backtesting engine"""
gateway_name: str = "BACKTESTING"
def __init__(self, lab: AlphaLab) -> None:
"""Constructor"""
self.lab: AlphaLab = lab
self.vt_symbols: list[str] = []
self.start: datetime
self.end: datetime
self.long_rates: dict[str, float] = {}
self.short_rates: dict[str, float] = {}
self.sizes: dict[str, float] = {}
self.priceticks: dict[str, float] = {}
self.capital: float = 0
self.risk_free: float = 0
self.annual_days: int = 0
self.strategy_class: type[AlphaStrategy]
self.strategy: AlphaStrategy
self.bars: dict[str, BarData] = {}
self.datetime: datetime | None = None
self.interval: Interval
self.history_data: dict[tuple, BarData] = {}
self.dts: set[datetime] = set()
self.limit_order_count: int = 0
self.limit_orders: dict[str, OrderData] = {}
self.active_limit_orders: dict[str, OrderData] = {}
self.trade_count: int = 0
self.trades: dict[str, TradeData] = {}
self.logs: list[str] = []
self.daily_results: dict[date, PortfolioDailyResult] = {}
self.daily_df: pl.DataFrame
self.pre_closes: defaultdict = defaultdict(float)
self.cash: float = 0
self.signal_df: pl.DataFrame
def set_parameters(
self,
vt_symbols: list[str],
interval: Interval,
start: datetime,
end: datetime,
capital: int = 1_000_000,
risk_free: float = 0,
annual_days: int = 240
) -> None:
"""Set parameters"""
self.vt_symbols = vt_symbols
self.interval = interval
self.start = start
self.end = end
self.capital = capital
self.risk_free = risk_free
self.annual_days = annual_days
self.cash = capital
contract_settings: dict = self.lab.load_contract_setttings()
for vt_symbol in vt_symbols:
setting: dict | None = contract_settings.get(vt_symbol, None)
if not setting:
logger.warning(f"找不到合约{vt_symbol}的交易配置，请检查！")
continue
self.long_rates[vt_symbol] = setting["long_rate"]
self.short_rates[vt_symbol] = setting["short_rate"]
self.sizes[vt_symbol] = setting["size"]
self.priceticks[vt_symbol] = setting["pricetick"]
def add_strategy(self, strategy_class: type, setting: dict, signal_df: pl.DataFrame) -> None:
"""Add strategy"""
self.strategy_class = strategy_class
self.strategy = strategy_class(
self, strategy_class.__name__, copy(self.vt_symbols), setting
)
self.signal_df = signal_df
def load_data(self) -> None:
"""Load historical data"""
logger.info("开始加载历史数据")
if not self.end:
self.end = datetime.now()
if self.start >= self.end:
logger.info("起始日期必须小于结束日期")
return
# Clear previously loaded historical data
self.history_data.clear()
self.dts.clear()
# Load historical data for each symbol
empty_symbols: list[str] = []
for vt_symbol in tqdm(self.vt_symbols, total=len(self.vt_symbols)):
data: list[BarData] = self.lab.load_bar_data(
vt_symbol,
self.interval,
self.start,
self.end
)
for bar in data:
self.dts.add(bar.datetime)
self.history_data[(bar.datetime, vt_symbol)] = bar
data_count = len(data)
if not data_count:
empty_symbols.append(vt_symbol)
if empty_symbols:
logger.info(f"部分合约历史数据为空：{empty_symbols}")
logger.info("所有历史数据加载完成")
def run_backtesting(self) -> None:
"""Start backtesting"""
self.strategy.on_init()
logger.info("策略初始化完成")
# Use remaining historical data for strategy backtesting
dts: list = list(self.dts)
dts.sort()
logger.info("开始回放历史数据")
for dt in dts:
try:
self.new_bars(dt)
except Exception:
logger.info("触发异常，回测终止")
logger.info(traceback.format_exc())
return
logger.info("历史数据回放结束")
def calculate_result(self) -> pl.DataFrame | None:
"""Calculate daily mark-to-market profit and loss"""
logger.info("开始计算逐日盯市盈亏")
if not self.trades:
logger.info("成交记录为空，无法计算")
return None
for trade in self.trades.values():
if not trade.datetime:
continue
d: date = trade.datetime.date()
daily_result: PortfolioDailyResult = self.daily_results[d]
daily_result.add_trade(trade)
pre_closes: dict[str, float] = {}
start_poses: dict[str, float] = {}
for daily_result in self.daily_results.values():
daily_result.calculate_pnl(
pre_closes,
start_poses,
self.sizes,
self.long_rates,
self.short_rates
)
pre_closes = daily_result.close_prices
start_poses = daily_result.end_poses
results: dict = defaultdict(list)
for daily_result in self.daily_results.values():
fields: list = [
"date", "trade_count", "turnover",
"commission", "trading_pnl",
"holding_pnl", "total_pnl", "net_pnl"
]
for key in fields:
value = getattr(daily_result, key)
results[key].append(value)
if results:
self.daily_df = pl.DataFrame([
pl.Series("date", results["date"], dtype=pl.Date),
pl.Series("trade_count", results["trade_count"], dtype=pl.Int64),
pl.Series("turnover", results["turnover"], dtype=pl.Float64),
pl.Series("commission", results["commission"], dtype=pl.Float64),
pl.Series("trading_pnl", results["trading_pnl"], dtype=pl.Float64),
pl.Series("holding_pnl", results["holding_pnl"], dtype=pl.Float64),
pl.Series("total_pnl", results["total_pnl"], dtype=pl.Float64),
pl.Series("net_pnl", results["net_pnl"], dtype=pl.Float64),
])
logger.info("逐日盯市盈亏计算完成")
return self.daily_df
def calculate_statistics(self) -> dict:
"""Calculate strategy statistics"""
logger.info("开始计算策略统计指标")
# Initialize statistics
start_date: str = ""
end_date: str = ""
total_days: int = 0
profit_days: int = 0
loss_days: int = 0
end_balance: float = 0
max_drawdown: float = 0
max_ddpercent: float = 0
max_drawdown_duration: int = 0
total_net_pnl: float = 0
daily_net_pnl: float = 0
total_commission: float = 0
daily_commission: float = 0
total_turnover: float = 0
daily_turnover: float = 0
total_trade_count: int = 0
daily_trade_count: float = 0
total_return: float = 0
annual_return: float = 0
daily_return: float = 0
return_std: float = 0
sharpe_ratio: float = 0
return_drawdown_ratio: float = 0
# Check if bankruptcy occurred
positive_balance: bool = False
# Calculate capital-related metrics
df: pl.DataFrame = self.daily_df
if df is not None:
df = df.with_columns(
# Strategy capital
balance=pl.col("net_pnl").cum_sum() + self.capital
).with_columns(
# Strategy return
pl.col("balance").pct_change().fill_null(0).alias("return"),
# Capital high watermark
highlevel=pl.col("balance").cum_max()
).with_columns(
# Capital drawdown
drawdown=pl.col("balance") - pl.col("highlevel"),
# Percentage drawdown
ddpercent=(pl.col("balance") / pl.col("highlevel") - 1) * 100
)
# Check if bankruptcy occurred
positive_balance = (df["balance"] > 0).all()
if not positive_balance:
logger.info("回测中出现爆仓（资金小于等于0），无法计算策略统计指标")
# Save data object
self.daily_df = df
# Calculate statistics
if positive_balance:
start_date = df["date"][0]
end_date = df["date"][-1]
total_days = len(df)
profit_days = df.filter(pl.col("net_pnl") > 0).height
loss_days = df.filter(pl.col("net_pnl") < 0).height
end_balance = df["balance"][-1]
max_drawdown = cast(float, df["drawdown"].min())
max_ddpercent = cast(float, df["ddpercent"].min())
max_drawdown_end_idx = cast(int, df["drawdown"].arg_min())
max_drawdown_end = df["date"][max_drawdown_end_idx]
if isinstance(max_drawdown_end, date):
max_drawdown_start_idx = cast(int, df.slice(0, max_drawdown_end_idx + 1)["balance"].arg_max())
max_drawdown_start = df["date"][max_drawdown_start_idx]
max_drawdown_duration = (max_drawdown_end - max_drawdown_start).days
else:
max_drawdown_duration = 0
total_net_pnl = df["net_pnl"].sum()
daily_net_pnl = total_net_pnl / total_days
total_commission = df["commission"].sum()
daily_commission = total_commission / total_days
total_turnover = df["turnover"].sum()
daily_turnover = total_turnover / total_days
total_trade_count = cast(int, df["trade_count"].sum())
daily_trade_count = total_trade_count / total_days
total_return = (end_balance / self.capital - 1) * 100
annual_return = total_return / total_days * self.annual_days
daily_return = cast(float, df["return"].mean()) * 100
return_std = cast(float, df["return"].std()) * 100
if return_std:
daily_risk_free = self.risk_free / np.sqrt(self.annual_days)
sharpe_ratio = (daily_return - daily_risk_free) / return_std * np.sqrt(self.annual_days)
else:
sharpe_ratio = 0
return_drawdown_ratio = -total_net_pnl / max_drawdown
# Output results
logger.info("-" * 30)
logger.info(f"首个交易日： {start_date}")
logger.info(f"最后交易日： {end_date}")
logger.info(f"总交易日： {total_days}")
logger.info(f"盈利交易日： {profit_days}")
logger.info(f"亏损交易日： {loss_days}")
logger.info(f"起始资金： {self.capital:,.2f}")
logger.info(f"结束资金： {end_balance:,.2f}")
logger.info(f"总收益率： {total_return:,.2f}%")
logger.info(f"年化收益： {annual_return:,.2f}%")
logger.info(f"最大回撤: {max_drawdown:,.2f}")
logger.info(f"百分比最大回撤: {max_ddpercent:,.2f}%")
logger.info(f"最长回撤天数: {max_drawdown_duration}")
logger.info(f"总盈亏： {total_net_pnl:,.2f}")
logger.info(f"总手续费： {total_commission:,.2f}")
logger.info(f"总成交金额： {total_turnover:,.2f}")
logger.info(f"总成交笔数： {total_trade_count}")
logger.info(f"日均盈亏： {daily_net_pnl:,.2f}")
logger.info(f"日均手续费： {daily_commission:,.2f}")
logger.info(f"日均成交金额： {daily_turnover:,.2f}")
logger.info(f"日均成交笔数： {daily_trade_count}")
logger.info(f"日均收益率： {daily_return:,.2f}%")
logger.info(f"收益标准差： {return_std:,.2f}%")
logger.info(f"Sharpe Ratio： {sharpe_ratio:,.2f}")
logger.info(f"收益回撤比： {return_drawdown_ratio:,.2f}")
statistics: dict = {
"start_date": start_date,
"end_date": end_date,
"total_days": total_days,
"profit_days": profit_days,
"loss_days": loss_days,
"capital": self.capital,
"end_balance": end_balance,
"max_drawdown": max_drawdown,
"max_ddpercent": max_ddpercent,
"max_drawdown_duration": max_drawdown_duration,
"total_net_pnl": total_net_pnl,
"daily_net_pnl": daily_net_pnl,
"total_commission": total_commission,
"daily_commission": daily_commission,
"total_turnover": total_turnover,
"daily_turnover": daily_turnover,
"total_trade_count": total_trade_count,
"daily_trade_count": daily_trade_count,
"total_return": total_return,
"annual_return": annual_return,
"daily_return": daily_return,
"return_std": return_std,
"sharpe_ratio": sharpe_ratio,
"return_drawdown_ratio": return_drawdown_ratio,
}
# Filter extreme values
for key, value in statistics.items():
if value in (np.inf, -np.inf):
value = 0
statistics[key] = np.nan_to_num(value)
logger.info("策略统计指标计算完成")
return statistics
def show_chart(self) -> None:
"""Display chart"""
df: pl.DataFrame = self.daily_df
fig = make_subplots(
rows=4,
cols=1,
subplot_titles=["Balance", "Drawdown", "Daily Pnl", "Pnl Distribution"],
vertical_spacing=0.06
)
balance_line = go.Scatter(
x=df["date"],
y=df["balance"],
mode="lines",
name="Balance"
)
drawdown_scatter = go.Scatter(
x=df["date"],
y=df["drawdown"],
fillcolor="red",
fill='tozeroy',
mode="lines",
name="Drawdown"
)
pnl_bar = go.Bar(y=df["net_pnl"], name="Daily Pnl")
pnl_histogram = go.Histogram(x=df["net_pnl"], nbinsx=100, name="Days")
fig.add_trace(balance_line, row=1, col=1)
fig.add_trace(drawdown_scatter, row=2, col=1)
fig.add_trace(pnl_bar, row=3, col=1)
fig.add_trace(pnl_histogram, row=4, col=1)
fig.update_layout(height=1000, width=1000)
fig.show()
def show_performance(self, benchmark_symbol: str) -> None:
"""Display performance metrics"""
# Load benchmark prices
benchmark_bars: list[BarData] = self.lab.load_bar_data(benchmark_symbol, self.interval, self.start, self.end)
benchmark_prices: list[float] = []
for bar in benchmark_bars:
benchmark_prices.append(bar.close_price)
# Calculate strategy performance
performance_df: pl.DataFrame = (
self.daily_df.with_columns(
# Cumulative return
cumulative_return=pl.col("balance").pct_change().cum_sum(),
# Cumulative cost
cumulative_cost=(pl.col("commission") / pl.col("balance").shift(1)).cum_sum()
).with_columns(
# Benchmark price
benchmark_price=pl.Series(values=benchmark_prices, dtype=pl.Float64)
).with_columns(
# Benchmark return
benchmark_return=pl.col("benchmark_price").pct_change().cum_sum()
).with_columns(
# Excess return
excess_return=(pl.col("cumulative_return") - pl.col("benchmark_return"))
).with_columns(
# Net excess return
net_excess_return=(pl.col("excess_return") - pl.col("cumulative_cost")),
).with_columns(
# Excess return drawdown
excess_return_drawdown=(pl.col("excess_return") - pl.col("excess_return").cum_max()),
# Net excess return drawdown
net_excess_return_drawdown=(pl.col("net_excess_return") - pl.col("net_excess_return").cum_max())
)
)
# Draw chart
fig: go.Figure = make_subplots(
rows=5,
cols=1,
subplot_titles=["Return", "Alpha", "Turnover", "Alpha Drawdown", "Alpha Drawdown with Cost"],
vertical_spacing=0.06
)
strategy_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["cumulative_return"],
mode="lines",
name="Strategy"
)
net_strategy_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["cumulative_return"] - performance_df["cumulative_cost"],
mode="lines",
name="Strategy with Cost"
)
benchmark_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["benchmark_return"],
mode="lines",
name="Benchmark"
)
excess_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["excess_return"],
mode="lines",
name="Alpha"
)
net_excess_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["net_excess_return"],
mode="lines",
name="Alpha with Cost"
)
turnover_curve: go.Scatter = go.Scatter(
x=self.daily_df["date"],
y=self.daily_df["turnover"] / self.daily_df["balance"].shift(1),
name="Turnover",
)
excess_drawdown_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["excess_return_drawdown"],
fill='tozeroy',
mode="lines",
name="Alpha Drawdown"
)
net_excess_drawdown_curve: go.Scatter = go.Scatter(
x=performance_df["date"],
y=performance_df["net_excess_return_drawdown"],
fill='tozeroy',
mode="lines",
name="Alpha Drawdown with Cost"
)
fig.add_trace(strategy_curve, row=1, col=1)
fig.add_trace(net_strategy_curve, row=1, col=1)
fig.add_trace(benchmark_curve, row=1, col=1)
fig.add_trace(excess_curve, row=2, col=1)
fig.add_trace(net_excess_curve, row=2, col=1)
fig.add_trace(turnover_curve, row=3, col=1)
fig.add_trace(excess_drawdown_curve, row=4, col=1)
fig.add_trace(net_excess_drawdown_curve, row=5, col=1)
fig.update_layout(
height=1500,
width=1200,
plot_bgcolor="white",
paper_bgcolor="white",
xaxis=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
xaxis2=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
xaxis3=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
xaxis4=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
xaxis5=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
yaxis=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
yaxis2=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
yaxis3=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
yaxis4=dict(showgrid=True, gridwidth=1, gridcolor='LightGray'),
yaxis5=dict(showgrid=True, gridwidth=1, gridcolor='LightGray')
)
fig.show()
def update_daily_close(self, bars: dict[str, BarData], dt: datetime) -> None:
"""Update daily closing price"""
d: date = dt.date()
close_prices: dict[str, float] = {}
for bar in bars.values():
if not bar.close_price:
close_prices[bar.vt_symbol] = self.pre_closes[bar.vt_symbol]
else:
close_prices[bar.vt_symbol] = bar.close_price
daily_result: PortfolioDailyResult | None = self.daily_results.get(d, None)
if daily_result:
daily_result.update_close_prices(close_prices)
else:
self.daily_results[d] = PortfolioDailyResult(d, close_prices)
def new_bars(self, dt: datetime) -> None:
"""Push historical data"""
self.datetime = dt
bars: dict[str, BarData] = {}
for vt_symbol in self.vt_symbols:
last_bar = self.bars.get(vt_symbol, None)
if last_bar:
if last_bar.close_price:
self.pre_closes[vt_symbol] = last_bar.close_price
bar: BarData | None = self.history_data.get((dt, vt_symbol), None)
# Check if historical data for the specified time of the contract is obtained
if bar:
# Update K-line for order matching
self.bars[vt_symbol] = bar
# Cache K-line data for strategy.on_bars update
bars[vt_symbol] = bar
# If not available, but there is contract data cached in the self.bars dictionary, use previous data to fill
elif vt_symbol in self.bars:
old_bar: BarData = self.bars[vt_symbol]
fill_bar: BarData = BarData(
symbol=old_bar.symbol,
exchange=old_bar.exchange,
datetime=dt,
open_price=old_bar.close_price,
high_price=old_bar.close_price,
low_price=old_bar.close_price,
close_price=old_bar.close_price,
gateway_name=old_bar.gateway_name
)
self.bars[vt_symbol] = fill_bar
self.cross_order()
self.strategy.on_bars(bars)
self.update_daily_close(self.bars, dt)
def cross_order(self) -> None:
"""Match limit orders"""
for order in list(self.active_limit_orders.values()):
bar: BarData = self.bars[order.vt_symbol]
long_cross_price: float = bar.low_price
short_cross_price: float = bar.high_price
long_best_price: float = bar.open_price
short_best_price: float = bar.open_price
# Push order status update for unfilled orders
if order.status == Status.SUBMITTING:
order.status = Status.NOTTRADED
self.strategy.update_order(order)
# Calculate price limits
pricetick: float = self.priceticks[order.vt_symbol]
pre_close: float = self.pre_closes.get(order.vt_symbol, 0)
limit_up: float = round_to(pre_close * 1.1, pricetick)
limit_down: float = round_to(pre_close * 0.9, pricetick)
# Check limit orders that can be matched
long_cross: bool = (
order.direction == Direction.LONG
and order.price >= long_cross_price
and long_cross_price > 0
and bar.low_price < limit_up # Not a full-day limit-up market
)
short_cross: bool = (
order.direction == Direction.SHORT
and order.price <= short_cross_price
and short_cross_price > 0
and bar.high_price > limit_down # Not a full-day limit-down market
)
if not long_cross and not short_cross:
continue
# Push order status update for filled orders
order.traded = order.volume
order.status = Status.ALLTRADED
self.strategy.update_order(order)
if order.vt_orderid in self.active_limit_orders:
self.active_limit_orders.pop(order.vt_orderid)
# Generate trade information
self.trade_count += 1
if long_cross:
trade_price = min(order.price, long_best_price)
else:
trade_price = max(order.price, short_best_price)
trade: TradeData = TradeData(
symbol=order.symbol,
exchange=order.exchange,
orderid=order.orderid,
tradeid=str(self.trade_count),
direction=order.direction,
offset=order.offset,
price=trade_price,
volume=order.volume,
datetime=self.datetime,
gateway_name=self.gateway_name,
)
# Update available funds
size: float = self.sizes[trade.vt_symbol]
trade_turnover: float = trade.price * trade.volume * size
if trade.direction == Direction.LONG:
trade_commission: float = trade_turnover * self.long_rates[trade.vt_symbol]
else:
trade_commission = trade_turnover * self.short_rates[trade.vt_symbol]
if trade.direction == Direction.LONG:
self.cash -= trade_turnover
else:
self.cash += trade_turnover
self.cash -= trade_commission
# Push trade information
self.strategy.update_trade(trade)
self.trades[trade.vt_tradeid] = trade
def get_signal(self) -> pl.DataFrame:
"""Get model prediction signal for current time"""
if not self.datetime:
self.write_log("尚未开始数据回放，无法加载模型预测值")
return pl.DataFrame()
dt: datetime = self.datetime.replace(tzinfo=None)
signal: pl.DataFrame = self.signal_df.filter(pl.col("datetime") == dt)
if signal.is_empty():
self.write_log(f"找不到{dt}对应的信号模型预测值")
return signal
def send_order(
self,
strategy: AlphaStrategy,
vt_symbol: str,
direction: Direction,
offset: Offset,
price: float,
volume: float,
) -> list[str]:
"""Send order"""
price = round_to(price, self.priceticks[vt_symbol])
symbol, exchange = extract_vt_symbol(vt_symbol)
self.limit_order_count += 1
order: OrderData = OrderData(
symbol=symbol,
exchange=exchange,
orderid=str(self.limit_order_count),
direction=direction,
offset=offset,
price=price,
volume=volume,
status=Status.SUBMITTING,
datetime=self.datetime,
gateway_name=self.gateway_name,
)
self.active_limit_orders[order.vt_orderid] = order
self.limit_orders[order.vt_orderid] = order
return [order.vt_orderid]
def cancel_order(self, strategy: AlphaStrategy, vt_orderid: str) -> None:
"""Cancel order"""
if vt_orderid not in self.active_limit_orders:
return
order: OrderData = self.active_limit_orders.pop(vt_orderid)
order.status = Status.CANCELLED
self.strategy.update_order(order)
def write_log(self, msg: str, strategy: AlphaStrategy | None = None) -> None:
"""Output log message"""
msg = f"{self.datetime} {msg}"
self.logs.append(msg)
def get_all_trades(self) -> list[TradeData]:
"""Get all trade information"""
return list(self.trades.values())
def get_all_orders(self) -> list[OrderData]:
"""Get all order information"""
return list(self.limit_orders.values())
def get_all_daily_results(self) -> list["PortfolioDailyResult"]:
"""Get all daily profit and loss information"""
return list(self.daily_results.values())
def get_cash_available(self) -> float:
"""Get current available cash"""
return self.cash
def get_holding_value(self) -> float:
"""Get current holding market value"""
holding_value: float = 0
for vt_symbol, pos in self.strategy.pos_data.items():
bar: BarData = self.bars[vt_symbol]
size: float = self.sizes[vt_symbol]
holding_value += bar.close_price * pos * size
return holding_value
class ContractDailyResult:
"""Contract daily profit and loss result"""
def __init__(self, result_date: date, close_price: float) -> None:
"""Constructor"""
self.date: date = result_date
self.close_price: float = close_price
self.pre_close: float = 0
self.trades: list[TradeData] = []
self.trade_count: int = 0
self.start_pos: float = 0
self.end_pos: float = 0
self.turnover: float = 0
self.commission: float = 0
self.trading_pnl: float = 0
self.holding_pnl: float = 0
self.total_pnl: float = 0
self.net_pnl: float = 0
def add_trade(self, trade: TradeData) -> None:
"""Add trade information"""
self.trades.append(trade)
def calculate_pnl(
self,
pre_close: float,
start_pos: float,
size: float,
long_rate: float,
short_rate: float
) -> None:
"""Calculate profit and loss"""
# If there is no previous close price, use 1 instead to avoid division error
if pre_close:
self.pre_close = pre_close
# else:
# self.pre_close = 1
# Calculate holding profit and loss
self.start_pos = start_pos
self.end_pos = start_pos
self.holding_pnl = self.start_pos * (self.close_price - self.pre_close) * size
# Calculate trading profit and loss
self.trade_count = len(self.trades)
for trade in self.trades:
if trade.direction == Direction.LONG:
pos_change: float = trade.volume
rate: float = long_rate
else:
pos_change = -trade.volume
rate = short_rate
self.end_pos += pos_change
turnover: float = trade.volume * size * trade.price
self.trading_pnl += pos_change * (self.close_price - trade.price) * size
self.turnover += turnover
self.commission += turnover * rate
# Calculate daily profit and loss
self.total_pnl = self.trading_pnl + self.holding_pnl
self.net_pnl = self.total_pnl - self.commission
def update_close_price(self, close_price: float) -> None:
"""Update daily close price"""
self.close_price = close_price
class PortfolioDailyResult:
"""Portfolio daily profit and loss result"""
def __init__(self, result_date: date, close_prices: dict[str, float]) -> None:
"""Constructor"""
self.date: date = result_date
self.close_prices: dict[str, float] = close_prices
self.pre_closes: dict[str, float] = {}
self.start_poses: dict[str, float] = {}
self.end_poses: dict[str, float] = {}
self.contract_results: dict[str, ContractDailyResult] = {}
for vt_symbol, close_price in close_prices.items():
self.contract_results[vt_symbol] = ContractDailyResult(result_date, close_price)
self.trade_count: int = 0
self.turnover: float = 0
self.commission: float = 0
self.trading_pnl: float = 0
self.holding_pnl: float = 0
self.total_pnl: float = 0
self.net_pnl: float = 0
def add_trade(self, trade: TradeData) -> None:
"""Add trade information"""
contract_result: ContractDailyResult = self.contract_results[trade.vt_symbol]
contract_result.add_trade(trade)
def calculate_pnl(
self,
pre_closes: dict[str, float],
start_poses: dict[str, float],
sizes: dict[str, float],
long_rates: dict[str, float],
short_rates: dict[str, float]
) -> None:
"""Calculate profit and loss"""
self.pre_closes = pre_closes
self.start_poses = start_poses
for vt_symbol, contract_result in self.contract_results.items():
contract_result.calculate_pnl(
pre_closes.get(vt_symbol, 0),
start_poses.get(vt_symbol, 0),
sizes[vt_symbol],
long_rates[vt_symbol],
short_rates[vt_symbol]
)
self.trade_count += contract_result.trade_count
self.turnover += contract_result.turnover
self.commission += contract_result.commission
self.trading_pnl += contract_result.trading_pnl
self.holding_pnl += contract_result.holding_pnl
self.total_pnl += contract_result.total_pnl
self.net_pnl += contract_result.net_pnl
self.end_poses[vt_symbol] = contract_result.end_pos
def update_close_prices(self, close_prices: dict[str, float]) -> None:
"""Update daily close prices"""
self.close_prices.update(close_prices)
for vt_symbol, close_price in close_prices.items():
contract_result: ContractDailyResult | None = self.contract_results.get(vt_symbol, None)
if contract_result:
contract_result.update_close_price(close_price)
else:
self.contract_results[vt_symbol] = ContractDailyResult(self.date, close_price)