research/pairs_trading
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

trades performance analysis

Browse Source
This commit is contained in:
Oleg Sheynin 2025-07-30 17:08:06 +00:00
parent 566dd9bbdc
commit c1c72f46a6
3 changed files with 46 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
configuration/new_zscore.cfg
View File

@ -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",

133
lib/pt_strategy/results.py
View File

@ -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
@ -574,7 +527,3 @@ class PairResearchResult:
"""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

15
lib/pt_strategy/trading_strategy.py
View File

@ -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}")