From c1c72f46a6ad2b6b05774538098cfc2eed77cb8d Mon Sep 17 00:00:00 2001 From: Oleg Sheynin Date: Wed, 30 Jul 2025 17:08:06 +0000 Subject: [PATCH] trades performance analysis --- configuration/new_zscore.cfg | 4 +- lib/pt_strategy/results.py | 137 +++++++++------------------- lib/pt_strategy/trading_strategy.py | 15 +-- 3 files changed, 46 insertions(+), 110 deletions(-) diff --git a/configuration/new_zscore.cfg b/configuration/new_zscore.cfg index d97e9e1..2ea53a8 100644 --- a/configuration/new_zscore.cfg +++ b/configuration/new_zscore.cfg @@ -33,8 +33,8 @@ } # ====== End of Session Closeout ====== - # "close_outstanding_positions": true, - "close_outstanding_positions": false, + "close_outstanding_positions": true, + # "close_outstanding_positions": false, "trading_hours": { "timezone": "America/New_York", "begin_session": "7:30:00", diff --git a/lib/pt_strategy/results.py b/lib/pt_strategy/results.py index 474f574..c3624f2 100644 --- a/lib/pt_strategy/results.py +++ b/lib/pt_strategy/results.py @@ -195,61 +195,58 @@ def convert_timestamp(timestamp: Any) -> Optional[datetime]: DayT = str TradeT = Dict[str, Any] -OutstandingPositionT = List[Dict[str, Any]] +OutstandingPositionT = Dict[str, Any] class PairResearchResult: """ Class to handle pair research results for a single pair across multiple days. Simplified version of BacktestResult focused on single pair analysis. """ trades_: Dict[DayT, pd.DataFrame] - outstanding_positions_: Dict[DayT, OutstandingPositionT] + outstanding_positions_: Dict[DayT, List[OutstandingPositionT]] + symbol_roundtrip_trades_: Dict[str, List[Dict[str, Any]]] + def __init__(self, config: Dict[str, Any]) -> None: self.config_ = config self.trades_ = {} self.outstanding_positions_ = {} self.total_realized_pnl = 0.0 - self.symbol_roundtrip_trades_: Dict[str, List[Dict[str, Any]]] = {} + self.symbol_roundtrip_trades_ = {} def add_day_results(self, day: DayT, trades: pd.DataFrame, outstanding_positions: List[Dict[str, Any]]) -> None: assert isinstance(trades, pd.DataFrame) self.trades_[day] = trades self.outstanding_positions_[day] = outstanding_positions - @property - def all_trades(self) -> List[TradeT]: - """Get all trades across all days as a flat list.""" - all_trades_list = [] - for day_trades in self.trades_.values(): - all_trades_list.extend(day_trades) - return all_trades_list + # def all_trades(self) -> List[TradeT]: + # """Get all trades across all days as a flat list.""" + # all_trades_list: List[TradeT] = [] + # for day_trades in self.trades_.values(): + # all_trades_list.extend(day_trades.to_dict(orient="records")) + # return all_trades_list - @property def outstanding_positions(self) -> List[OutstandingPositionT]: """Get all outstanding positions across all days as a flat list.""" - all_positions = [] - for day_positions in self.outstanding_positions_.values(): - all_positions.extend(day_positions) - return all_positions - + res: List[Dict[str, Any]] = [] + for day in self.outstanding_positions_.keys(): + res.extend(self.outstanding_positions_[day]) + return res def calculate_returns(self) -> None: """Calculate and store total returns for the single pair across all days.""" - roundtrip_trades = self.extract_roundtrip_trades() + self.extract_roundtrip_trades() self.total_realized_pnl = 0.0 - for day, day_trades in roundtrip_trades.items(): + for day, day_trades in self.symbol_roundtrip_trades_.items(): for trade in day_trades: self.total_realized_pnl += trade['symbol_return'] - def extract_roundtrip_trades(self) -> Dict[str, List[Dict[str, Any]]]: + def extract_roundtrip_trades(self) -> None: """ Extract round-trip trades by day, grouping open/close pairs for each symbol. Returns a dictionary with day as key and list of completed round-trip trades. """ - roundtrip_trades_by_day = {} - def _symbol_return(trade1_side: str, trade1_px: float, trade2_side: str, trade2_px: float) -> float: if trade1_side == "BUY" and trade2_side == "SELL": return (trade2_px - trade1_px) / trade1_px * 100 @@ -260,11 +257,9 @@ class PairResearchResult: # Process each day separately for day, day_trades in self.trades_.items(): - if not day_trades or len(day_trades) < 4: - continue - + # Sort trades by timestamp for the day - sorted_trades = sorted(day_trades, key=lambda x: x["timestamp"] if x["timestamp"] else pd.Timestamp.min) + sorted_trades = day_trades #sorted(day_trades, key=lambda x: x["timestamp"] if x["timestamp"] else pd.Timestamp.min) day_roundtrips = [] @@ -273,10 +268,10 @@ class PairResearchResult: if idx + 3 >= len(sorted_trades): break - trade_a_1 = sorted_trades[idx] # Open A - trade_b_1 = sorted_trades[idx + 1] # Open B - trade_a_2 = sorted_trades[idx + 2] # Close A - trade_b_2 = sorted_trades[idx + 3] # Close B + trade_a_1 = sorted_trades.iloc[idx] # Open A + trade_b_1 = sorted_trades.iloc[idx + 1] # Open B + trade_a_2 = sorted_trades.iloc[idx + 2] # Close A + trade_b_2 = sorted_trades.iloc[idx + 3] # Close B # Validate trade sequence if not (trade_a_1["action"] == "OPEN" and trade_a_2["action"] == "CLOSE"): @@ -304,10 +299,10 @@ class PairResearchResult: "symbol": trade_a_1["symbol"], "open_side": trade_a_1["side"], "open_price": trade_a_1["price"], - "open_time": trade_a_1["timestamp"], + "open_time": trade_a_1["time"], "close_side": trade_a_2["side"], "close_price": trade_a_2["price"], - "close_time": trade_a_2["timestamp"], + "close_time": trade_a_2["time"], "symbol_return": symbol_a_return, "pair_return": pair_return, "shares": funding_per_position / trade_a_1["price"], @@ -321,10 +316,10 @@ class PairResearchResult: "symbol": trade_b_1["symbol"], "open_side": trade_b_1["side"], "open_price": trade_b_1["price"], - "open_time": trade_b_1["timestamp"], + "open_time": trade_b_1["time"], "close_side": trade_b_2["side"], "close_price": trade_b_2["price"], - "close_time": trade_b_2["timestamp"], + "close_time": trade_b_2["time"], "symbol_return": symbol_b_return, "pair_return": pair_return, "shares": funding_per_position / trade_b_1["price"], @@ -334,27 +329,20 @@ class PairResearchResult: }) if day_roundtrips: - roundtrip_trades_by_day[day] = day_roundtrips + self.symbol_roundtrip_trades_[day] = day_roundtrips - return roundtrip_trades_by_day def print_returns_by_day(self) -> None: """ Print detailed return information for each day, grouped by day. Shows individual symbol round-trips and daily totals. """ - roundtrip_trades = self.extract_roundtrip_trades() - - if not roundtrip_trades: - print("\n====== NO ROUND-TRIP TRADES FOUND ======") - return print("\n====== PAIR RESEARCH RETURNS BY DAY ======") total_return_all_days = 0.0 - for day in sorted(roundtrip_trades.keys()): - day_trades = roundtrip_trades[day] + for day, day_trades in sorted(self.symbol_roundtrip_trades_.items()): print(f"\n--- {day} ---") @@ -362,10 +350,10 @@ class PairResearchResult: pair_returns = [] # Group trades by pair (every 2 trades form a pair) - for i in range(0, len(day_trades), 2): - if i + 1 < len(day_trades): - trade_a = day_trades[i] - trade_b = day_trades[i + 1] + for idx in range(0, len(day_trades), 2): + if idx + 1 < len(day_trades): + trade_a = day_trades[idx] + trade_b = day_trades[idx + 1] # Print individual symbol results print(f" {trade_a['open_time'].time()}-{trade_a['close_time'].time()}") @@ -394,18 +382,16 @@ class PairResearchResult: print(f"\n====== TOTAL RETURN ACROSS ALL DAYS ======") print(f"Total Return: {total_return_all_days:+.2f}%") - print(f"Total Days: {len(roundtrip_trades)}") - if len(roundtrip_trades) > 0: - print(f"Average Daily Return: {total_return_all_days / len(roundtrip_trades):+.2f}%") + print(f"Total Days: {len(self.symbol_roundtrip_trades_)}") + if len(self.symbol_roundtrip_trades_) > 0: + print(f"Average Daily Return: {total_return_all_days / len(self.symbol_roundtrip_trades_):+.2f}%") def get_return_summary(self) -> Dict[str, Any]: """ Get a summary of returns across all days. Returns a dictionary with key metrics. """ - roundtrip_trades = self.extract_roundtrip_trades() - - if not roundtrip_trades: + if len(self.symbol_roundtrip_trades_) == 0: return { "total_return": 0.0, "total_days": 0, @@ -420,7 +406,7 @@ class PairResearchResult: total_return = 0.0 total_pairs = 0 - for day, day_trades in roundtrip_trades.items(): + for day, day_trades in self.symbol_roundtrip_trades_.items(): day_return = 0.0 day_pairs = len(day_trades) // 2 # Each pair has 2 symbol trades @@ -439,38 +425,14 @@ class PairResearchResult: return { "total_return": total_return, - "total_days": len(roundtrip_trades), + "total_days": len(self.symbol_roundtrip_trades_), "total_pairs": total_pairs, - "average_daily_return": total_return / len(roundtrip_trades) if roundtrip_trades else 0.0, + "average_daily_return": total_return / len(self.symbol_roundtrip_trades_) if self.symbol_roundtrip_trades_ else 0.0, "best_day": best_day, "worst_day": worst_day, "daily_returns": daily_returns } - def print_single_day_results(self) -> None: - """Print results for all processed days.""" - all_trades_list = self.all_trades - if not all_trades_list: - print("No trades found.") - return - - print(f"Total trades processed: {len(all_trades_list)}") - - # Group trades by day - trades_by_day: Dict[str, List[Dict[str, Any]]] = {} - for trade in all_trades_list: - if trade["timestamp"]: - day = trade["timestamp"].date() - if day not in trades_by_day: - trades_by_day[day] = [] - trades_by_day[day].append(trade) - - for day, day_trades in sorted(trades_by_day.items()): - print(f"\n--- {day} ---") - for trade in day_trades: - print(f" {trade['timestamp'].time() if trade['timestamp'] else 'N/A'}: " - f"{trade['symbol']} {trade['side']} {trade['action']} @ ${trade['price']:.2f} " - f"({trade['status']})") def print_grand_totals(self) -> None: """Print grand totals for the single pair analysis.""" @@ -504,19 +466,10 @@ class PairResearchResult: print(f"PAIR RESEARCH PERFORMANCE ANALYSIS") print(f"{'='*60}") - # Calculate returns first self.calculate_returns() - - # Print detailed returns by day self.print_returns_by_day() - - # Print outstanding positions if any self.print_outstanding_positions() - - # Print grand totals self.print_grand_totals() - - # Print additional analysis self._print_additional_metrics() def _print_additional_metrics(self) -> None: @@ -551,7 +504,7 @@ class PairResearchResult: def print_outstanding_positions(self) -> None: """Print outstanding positions for the single pair.""" - all_positions = self.outstanding_positions + all_positions: List[OutstandingPositionT] = self.outstanding_positions() if not all_positions: print("\n====== NO OUTSTANDING POSITIONS ======") return @@ -573,8 +526,4 @@ class PairResearchResult: def get_total_realized_pnl(self) -> float: """Get total realized PnL.""" return self.total_realized_pnl - - def get_outstanding_positions(self) -> List[Dict[str, Any]]: - """Get outstanding positions.""" - return self.outstanding_positions - + \ No newline at end of file diff --git a/lib/pt_strategy/trading_strategy.py b/lib/pt_strategy/trading_strategy.py index 6cd76c1..8dcf0b3 100644 --- a/lib/pt_strategy/trading_strategy.py +++ b/lib/pt_strategy/trading_strategy.py @@ -390,22 +390,9 @@ def main() -> None: outstanding_positions=pt_strategy.outstanding_positions(), ) - # ADD RESULTS ANALYSIS - results.calculate_returns() - results.print_single_day_results() - # Store results with day name as key - # filename = os.path.basename(day) - # all_results[filename] = { - # "trades": pt_strategy.trades_.copy(), - # "outstanding_positions": pt_strategy.outstanding_positions_.copy(), - # } + results.analyze_pair_performance() - # print(f"Successfully processed {filename}") - - results.calculate_returns() - results.print_grand_totals() - results.print_outstanding_positions() if args.result_db.upper() != "NONE": print(f"\nResults stored in database: {args.result_db}")