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 = cast(float, df["net_pnl"].sum()) daily_net_pnl = total_net_pnl / total_days total_commission = cast(float, df["commission"].sum()) daily_commission = total_commission / total_days total_turnover = cast(float, 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)