Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def simulate_pair(self, pair: int = None):
- """Simulate using the original (vectorized) method for a given pair index or all pairs if None."""
- if pair is None:
- results = []
- highest_pnl = None
- highest_idx = None
- for i, (s1, s2, alpha, beta) in enumerate(self.pairs):
- df = self.simulate_pair(i)
- trades_df = self._extract_trades_from_df(df)
- # --- Use per-bar PnL for summary ---
- total_pnl = df["pnl"].sum()
- sharpe = df["pnl"].mean() / df["pnl"].std() * np.sqrt(252) if df["pnl"].std() > 0 else 0
- self.pair_results[i] = {
- "spread": df["spread"],
- "df_trades": trades_df,
- "s1": s1,
- "s2": s2,
- "alpha": alpha,
- "beta": beta,
- "pnl": total_pnl,
- "df": df,
- }
- results.append({"pair": f"{s1}-{s2}", "pnl": total_pnl, "sharpe": sharpe})
- if (highest_pnl is None) or (total_pnl > highest_pnl):
- highest_pnl = total_pnl
- highest_idx = i
- self.highest_pnl_data = self.pair_results[i]
- self.highest_pnl_pair_idx = i
- self.highest_pnl = highest_pnl
- self.last_mode = "regular"
- # Print summary table
- rpt = pd.DataFrame(results).sort_values("pnl", ascending=False).reset_index(drop=True)
- print("\n=== Backtest Results ===")
- print(rpt.to_string(index=False, float_format="%.2f"))
- return
- # --- Vectorized logic for a single pair ---
- s1, s2, alpha, beta = self.pairs[pair]
- px1 = self.prices[s1]
- px2 = self.prices[s2]
- window = deque(maxlen=60)
- prev_spread = None
- prev_sig = 0
- results = []
- for t, (p1, p2) in enumerate(zip(px1, px2)):
- spread = p1 - (alpha + beta * p2)
- window.append(spread)
- if len(window) < window.maxlen:
- results.append((px1.index[t], spread, 0, 0.0))
- prev_spread = spread
- continue
- mu, std = np.mean(window), np.std(window, ddof=0)
- z = (spread - mu) / std
- sig = prev_sig
- if z > self.entry_z:
- sig = -1
- elif z < -self.entry_z:
- sig = +1
- elif prev_sig != 0 and abs(z) <= self.exit_z:
- sig = 0
- pnl = prev_sig * (spread - prev_spread) * self.trade_size
- results.append((px1.index[t], spread, sig, pnl))
- prev_spread = spread
- prev_sig = sig
- df = pd.DataFrame(results, columns=["time","spread","signal","pnl"]).set_index("time")
- return df
Advertisement
Add Comment
Please, Sign In to add comment