p1otek/btc-trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement backtesting environment and refine strategy logic (v1.7.5)

Browse Source 
- Extract strategy logic into a reusable PingPongStrategy class for bot and backtester.
- Create src/strategies/backtest_engine.py for local historical testing using DB data.
- Add BACKTESTING_GUIDE.md with instructions on how to use the new environment.
- Update dashboard to show Net Realized PnL (including fees).
- Verified bot and backtester compatibility with existing Docker setup.
This commit is contained in:
Gemini CLI
2026-03-08 19:56:53 +01:00
parent f544b06753
commit 56d0237bbf
3 changed files with 287 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
src/strategies/ping_pong_bot.py
View File

@ -84,12 +84,93 @@ class DatabaseManager:
logger.error(f"DB Query Error for {symbol} {interval}: {e}")
return []
class PingPongStrategy:
"""Core Strategy Logic for Ping-Pong Scalping"""
def __init__(self, config):
self.config = config
self.direction = config.get('direction', 'long')
def rma(self, series, length):
alpha = 1 / length
return series.ewm(alpha=alpha, adjust=False).mean()
def calculate_indicators(self, df):
# RSI
rsi_cfg = self.config['rsi']
delta = df['close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
df['rsi'] = 100 - (100 / (1 + (self.rma(gain, rsi_cfg['period']) / self.rma(loss, rsi_cfg['period']))))
# Hurst
hurst_cfg = self.config['hurst']
mcl = hurst_cfg['period'] / 2
mcl_2 = int(round(mcl / 2))
df['tr'] = np.maximum(df['high'] - df['low'], np.maximum(abs(df['high'] - df['close'].shift(1)), abs(df['low'] - df['close'].shift(1))))
df['ma_mcl'] = self.rma(df['close'], mcl)
df['atr_mcl'] = self.rma(df['tr'], mcl)
df['center'] = df['ma_mcl'].shift(mcl_2).fillna(df['ma_mcl'])
mcm_off = hurst_cfg['multiplier'] * df['atr_mcl']
df['hurst_upper'] = df['center'] + mcm_off
df['hurst_lower'] = df['center'] - mcm_off
return df
def check_signals(self, df):
if len(df) < 3: return None
# finished = candle that just closed (e.g. 10:30)
# prev = candle before that (e.g. 10:29)
finished = df.iloc[-2]
prev = df.iloc[-3]
rsi_cfg, hurst_cfg = self.config['rsi'] or {}, self.config['hurst'] or {}
def is_crossing_up(p_val, p_band, c_open, c_close, c_band):
# 1. Crossed up BETWEEN candles
between = p_val < p_band and c_close >= c_band
# 2. Crossed up WITHIN this candle
within = c_open is not None and c_open < c_band and c_close >= c_band
return between or within
def is_crossing_down(p_val, p_band, c_open, c_close, c_band):
# 1. Crossed down BETWEEN candles
between = p_val > p_band and c_close <= c_band
# 2. Crossed down WITHIN this candle
within = c_open is not None and c_open > c_band and c_close <= c_band
return between or within
# Hurst Signals
h_upper_cross_down = is_crossing_down(prev['close'], prev['hurst_upper'], finished['open'], finished['close'], finished['hurst_upper'])
h_lower_cross_down = is_crossing_down(prev['close'], prev['hurst_lower'], finished['open'], finished['close'], finished['hurst_lower'])
# RSI Signals
rsi_cross_up = is_crossing_up(prev['rsi'], rsi_cfg.get('oversold', 30), None, finished['rsi'], rsi_cfg.get('oversold', 30))
rsi_cross_down = is_crossing_down(prev['rsi'], rsi_cfg.get('overbought', 70), None, finished['rsi'], rsi_cfg.get('overbought', 70))
l_open = (rsi_cfg.get('enabled_for_open') and rsi_cross_up) or \
(hurst_cfg.get('enabled_for_open') and h_lower_cross_down)
l_close = (rsi_cfg.get('enabled_for_close') and rsi_cross_down) or \
(hurst_cfg.get('enabled_for_close') and h_upper_cross_down)
s_open = (rsi_cfg.get('enabled_for_open') and rsi_cross_down) or \
(hurst_cfg.get('enabled_for_open') and h_upper_cross_down)
s_close = (rsi_cfg.get('enabled_for_close') and rsi_cross_up) or \
(hurst_cfg.get('enabled_for_close') and h_lower_cross_down)
if self.direction == 'long':
return "open" if l_open else ("close" if l_close else None)
else:
return "open" if s_open else ("close" if s_close else None)
class PingPongBot:
def __init__(self, config_path="config/ping_pong_config.yaml"):
self.version = "1.7.3"
self.version = "1.7.5"
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.strategy = PingPongStrategy(self.config)
# Explicitly load from ENV to ensure they are available
self.api_key = os.getenv("BYBIT_API_KEY") or os.getenv("API_KEY")
self.api_secret = os.getenv("BYBIT_API_SECRET") or os.getenv("API_SECRET")
@ -201,30 +282,8 @@ class PingPongBot:
except Exception as e:
logger.error(f"Failed to write to CSV log: {e}")
def rma(self, series, length):
alpha = 1 / length
return series.ewm(alpha=alpha, adjust=False).mean()
def calculate_indicators(self, df):
# RSI
rsi_cfg = self.config['rsi']
delta = df['close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
df['rsi'] = 100 - (100 / (1 + (self.rma(gain, rsi_cfg['period']) / self.rma(loss, rsi_cfg['period']))))
# Hurst
hurst_cfg = self.config['hurst']
mcl = hurst_cfg['period'] / 2
mcl_2 = int(round(mcl / 2))
df['tr'] = np.maximum(df['high'] - df['low'], np.maximum(abs(df['high'] - df['close'].shift(1)), abs(df['low'] - df['close'].shift(1))))
df['ma_mcl'] = self.rma(df['close'], mcl)
df['atr_mcl'] = self.rma(df['tr'], mcl)
df['center'] = df['ma_mcl'].shift(mcl_2).fillna(df['ma_mcl'])
mcm_off = hurst_cfg['multiplier'] * df['atr_mcl']
df['hurst_upper'] = df['center'] + mcm_off
df['hurst_lower'] = df['center'] - mcm_off
df = self.strategy.calculate_indicators(df)
last_row = df.iloc[-1]
now_str = datetime.now().strftime("%H:%M:%S")
self.current_indicators["rsi"] = {"value": float(last_row['rsi']), "timestamp": now_str}
@ -262,6 +321,7 @@ class PingPongBot:
await self.close_all_positions()
self.direction = new_direction
self.strategy.direction = new_direction
if self.direction == "long":
self.category = "inverse"
self.symbol = f"{self.base_coin}USD"
@ -396,52 +456,7 @@ class PingPongBot:
logger.error(f"Exchange Sync Error: {e}")
def check_signals(self, df):
if len(df) < 3: return None
# finished = candle that just closed (e.g. 10:30)
# prev = candle before that (e.g. 10:29)
finished = df.iloc[-2]
prev = df.iloc[-3]
rsi_cfg, hurst_cfg = self.config['rsi'] or {}, self.config['hurst'] or {}
def is_crossing_up(p_val, p_band, c_open, c_close, c_band):
# 1. Crossed up BETWEEN candles
between = p_val < p_band and c_close >= c_band
# 2. Crossed up WITHIN this candle
within = c_open is not None and c_open < c_band and c_close >= c_band
return between or within
def is_crossing_down(p_val, p_band, c_open, c_close, c_band):
# 1. Crossed down BETWEEN candles
between = p_val > p_band and c_close <= c_band
# 2. Crossed down WITHIN this candle
within = c_open is not None and c_open > c_band and c_close <= c_band
return between or within
# Hurst Signals - Only using 'is_crossing_down' as requested
h_upper_cross_down = is_crossing_down(prev['close'], prev['hurst_upper'], finished['open'], finished['close'], finished['hurst_upper'])
h_lower_cross_down = is_crossing_down(prev['close'], prev['hurst_lower'], finished['open'], finished['close'], finished['hurst_lower'])
# RSI Signals
rsi_cross_up = is_crossing_up(prev['rsi'], rsi_cfg.get('oversold', 30), None, finished['rsi'], rsi_cfg.get('oversold', 30))
rsi_cross_down = is_crossing_down(prev['rsi'], rsi_cfg.get('overbought', 70), None, finished['rsi'], rsi_cfg.get('overbought', 70))
l_open = (rsi_cfg.get('enabled_for_open') and rsi_cross_up) or \
(hurst_cfg.get('enabled_for_open') and h_lower_cross_down)
l_close = (rsi_cfg.get('enabled_for_close') and rsi_cross_down) or \
(hurst_cfg.get('enabled_for_close') and h_upper_cross_down)
s_open = (rsi_cfg.get('enabled_for_open') and rsi_cross_down) or \
(hurst_cfg.get('enabled_for_open') and h_upper_cross_down)
s_close = (rsi_cfg.get('enabled_for_close') and rsi_cross_up) or \
(hurst_cfg.get('enabled_for_close') and h_lower_cross_down)
if self.direction == 'long':
return "open" if l_open else ("close" if l_close else None)
else:
return "open" if s_open else ("close" if s_close else None)
return self.strategy.check_signals(df)
async def execute_trade(self, signal):
if not signal or not self.market_price: return
@ -544,11 +559,14 @@ class PingPongBot:
pnl_color = "green" if self.session_pnl >= 0 else "red"
pnl_btc_color = "green" if self.session_pnl_btc >= 0 else "red"
net_realized_pnl = self.total_realized_pnl - self.total_fees
cfg_table.add_row("Running Time", runtime_str)
cfg_table.add_row("Session PnL (USD)", f"[bold {pnl_color}]{'$' if self.session_pnl >= 0 else '-$'}{abs(self.session_pnl):.2f}[/]")
cfg_table.add_row("Session PnL (BTC)", f"[bold {pnl_btc_color}]{'{:+.6f}'.format(self.session_pnl_btc)} BTC[/]")
cfg_table.add_row("Total Fees", f"[bold red]-${self.total_fees:.2f}[/]")
cfg_table.add_row("Realized PnL", f"[bold {'green' if self.total_realized_pnl >= 0 else 'red'}]${self.total_realized_pnl:.2f}[/]")
cfg_table.add_row("Gross Realized PnL", f"[bold {'green' if self.total_realized_pnl >= 0 else 'red'}]${self.total_realized_pnl:.2f}[/]")
cfg_table.add_row("Net Realized PnL", f"[bold {'green' if net_realized_pnl >= 0 else 'red'}]${net_realized_pnl:.2f}[/]")
ind_table = Table(title="INDICATORS", box=box.ROUNDED, expand=True)
ind_table.add_column("Indicator"); ind_table.add_column("Value"); ind_table.add_column("Updated")