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

feat: add local PC support and setup instructions for backtesting (v1.7.8)

Browse Source
This commit is contained in:
Gemini CLI
2026-03-08 20:12:16 +01:00
parent 56d0237bbf
commit f3b186b01d
2 changed files with 176 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
BACKTESTING_GUIDE.md
View File

@ -10,7 +10,23 @@ To run the backtester, use the following command:
docker exec -it btc_ping_pong_bot python src/strategies/backtest_engine.py
```
## 2. Backtest Engine (`backtest_engine.py`)
## 2. Local PC Setup (Recommended for Memory Savings)
Running the backtest on your local machine is much faster and saves server memory.
### Steps:
1. **Clone/Sync your repo** to your local machine.
2. **Install dependencies**:
```bash
pip install -r requirements.txt
```
3. **Configure `.env`**: Ensure your local `.env` file has the correct `DB_HOST` (e.g., `20.20.20.20`).
4. **Run the engine**:
```bash
# From the project root
python3 src/strategies/backtest_engine.py --start_date 2024-01-01 --end_date 2024-01-31
```
## 3. Backtest Engine (`backtest_engine.py`)
The backtest engine reuses the core logic from `ping_pong_bot.py` via the `PingPongStrategy` class.
### Key Features:
@ -25,23 +41,35 @@ The backtest engine reuses the core logic from `ping_pong_bot.py` via the `PingP
docker exec -it btc_ping_pong_bot python src/strategies/backtest_engine.py
```
## 3. Strategy Optimization (Optional)
To find the absolute best parameters for RSI and Hurst, you can use **Optuna**.
## 3. Regime Testing (MA Switching)
You can test different Moving Average (MA) settings to see which regime detector works best for switching between `long` and `short` modes.
### Installation:
Inside the Docker container:
```bash
docker exec -it btc_ping_pong_bot pip install optuna
```
### Examples:
* **Test 15m SMA 200:**
```bash
docker exec -it btc_ping_pong_bot python src/strategies/backtest_engine.py --ma_period 200 --ma_interval 15m
```
* **Test 1h SMA 50:**
```bash
docker exec -it btc_ping_pong_bot python src/strategies/backtest_engine.py --ma_period 50 --ma_interval 1h
```
* **Test 4h SMA 100:**
```bash
docker exec -it btc_ping_pong_bot python src/strategies/backtest_engine.py --ma_period 100 --ma_interval 4h --limit 20000
```
### Planned Optimizer (`optimize_strategy.py`):
Once installed, we can implement an optimization script that searches for:
* `rsi_period`: 7 to 21
* `hurst_multiplier`: 1.2 to 2.5
* `partial_exit_pct`: 0.05 to 0.30
### How it works:
When `--ma_period` is provided, the engine:
1. Loads the MA timeframe data from the DB.
2. Merges it with the 1m price data.
3. Switches modes (`long` <=> `short`) whenever the 1m price crosses the MA.
4. **Automatically closes** the existing position on a mode switch, just like the live bot.
## 4. Local DB Data
The engine connects to your local PostgreSQL DB using the credentials in your `.env` file. It specifically queries the `candles` table for the symbol and interval defined in `config/ping_pong_config.yaml`.
## 4. Parameter Overrides
You can quickly override strategy settings without editing the config file:
* `--direction`: Force a specific mode (`long` or `short`).
* `--limit`: Change the number of 1m candles to test (default 10,000).
* `--config`: Use a different configuration file.
## 5. Interpreting Results
* **Final Equity:** Your simulated account balance after all trades.

183
src/strategies/backtest_engine.py
View File

@ -4,9 +4,14 @@ import yaml
import os
import asyncio
import asyncpg
import argparse
from datetime import datetime
from dotenv import load_dotenv
from ping_pong_bot import PingPongStrategy
# Load environment variables from .env
load_dotenv()
class BacktestEngine:
def __init__(self, config_path="config/ping_pong_config.yaml"):
with open(config_path, 'r') as f:
@ -25,14 +30,14 @@ class BacktestEngine:
# Settings
self.fee_rate = 0.0005 # 0.05% Taker
self.leverage = float(self.config.get('leverage_long' if self.direction == 'long' else 'leverage_short', 5.0))
self.leverage = 5.0 # Will be updated based on mode
self.pos_size_margin = float(self.config.get('pos_size_margin', 20.0))
self.partial_exit_pct = float(self.config.get('partial_exit_pct', 0.15))
self.trades = []
self.equity_curve = []
async def load_data(self, symbol, interval, limit=5000):
async def load_data(self, symbol, interval, limit=None, start_date=None, end_date=None):
conn = await asyncpg.connect(
host=os.getenv('DB_HOST', '20.20.20.20'),
port=int(os.getenv('DB_PORT', 5433)),
@ -40,66 +45,87 @@ class BacktestEngine:
password=os.getenv('DB_PASSWORD', ''),
database=os.getenv('DB_NAME', 'btc_data')
)
rows = await conn.fetch('''
SELECT time, open, high, low, close, volume
FROM candles
WHERE symbol = $1 AND interval = $2
ORDER BY time ASC LIMIT $3
''', symbol, interval, limit)
query = "SELECT time, open, high, low, close, volume FROM candles WHERE symbol = $1 AND interval = $2"
params = [symbol, interval]
if start_date:
query += f" AND time >= ${len(params)+1}"
params.append(datetime.fromisoformat(start_date))
if end_date:
query += f" AND time <= ${len(params)+1}"
params.append(datetime.fromisoformat(end_date))
query += " ORDER BY time ASC"
# Only use limit if NO dates are provided (to avoid truncating a specific range)
if limit and not (start_date or end_date):
query += f" LIMIT ${len(params)+1}"
params.append(limit)
rows = await conn.fetch(query, *params)
await conn.close()
df = pd.DataFrame([dict(r) for r in rows])
if df.empty: return df
df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
return df
def run(self, df):
def run(self, df, ma_df=None, ma_period=None):
if df.empty:
print("No data to run backtest.")
return
print(f"Starting backtest on {len(df)} candles...")
print(f"Period: {df.iloc[0]['time']} to {df.iloc[-1]['time']}")
df = self.strategy.calculate_indicators(df)
# Prepare MA for regime switching
ma_values = None
if ma_df is not None and ma_period:
ma_df['ma'] = ma_df['close'].rolling(window=ma_period).mean()
# Merge MA values into the main timeframe using 'ffill'
ma_subset = ma_df[['time', 'ma']].rename(columns={'time': 'ma_time'})
df = pd.merge_asof(df.sort_values('time'), ma_subset.sort_values('ma_time'),
left_on='time', right_on='ma_time', direction='backward')
ma_values = df['ma'].values
print(f"Regime Switching enabled (MA {ma_period} on {ma_df.iloc[0]['time'].strftime('%Y-%m-%d')} interval)")
# Start after enough candles for indicators
start_idx = max(self.config['rsi']['period'], self.config['hurst']['period']) + 5
start_idx = max(self.config['rsi']['period'], self.config['hurst']['period'], 100)
if start_idx >= len(df):
print(f"Error: Not enough candles for indicators. Need {start_idx}, got {len(df)}")
return
for i in range(start_idx, len(df)):
current_df = df.iloc[:i+1]
price = df.iloc[i]['close']
time = df.iloc[i]['time']
# 1. Regime Check (Dynamic Switch)
if ma_values is not None and not np.isnan(ma_values[i]):
new_direction = "long" if price > ma_values[i] else "short"
if new_direction != self.direction:
# Close existing position on regime change
if abs(self.position_size) > 0:
self.close_full_position(price, time, reason="Regime Switch")
self.direction = new_direction
self.strategy.direction = new_direction
# Update leverage based on mode
self.leverage = float(self.config.get('leverage_long' if self.direction == 'long' else 'leverage_short', 5.0))
# 2. Strategy Signal
signal = self.strategy.check_signals(current_df)
if signal == "open":
# Entry Logic
qty_usd = self.pos_size_margin * self.leverage
qty_btc = qty_usd / price
fee = qty_usd * self.fee_rate
self.balance -= fee
if self.direction == "long":
self.position_size += qty_btc
else: # Short
self.position_size -= qty_btc
self.entry_price = price # Simplified avg entry
self.trades.append({"time": time, "type": "Enter", "price": price, "fee": fee})
self.open_position(price, time)
elif signal == "close" and abs(self.position_size) > 0:
# Exit Logic
qty_btc_exit = abs(self.position_size) * self.partial_exit_pct
qty_usd_exit = qty_btc_exit * price
fee = qty_usd_exit * self.fee_rate
# Realized PnL
if self.direction == "long":
pnl = qty_btc_exit * (price - self.entry_price)
self.position_size -= qty_btc_exit
else: # Short
pnl = qty_btc_exit * (self.entry_price - price)
self.position_size += qty_btc_exit
self.balance += (pnl - fee)
self.trades.append({"time": time, "type": "Exit", "price": price, "pnl": pnl, "fee": fee})
self.close_partial_position(price, time)
# Mark to Market Equity
# 3. Mark to Market Equity
unrealized = 0
if self.direction == "long":
unrealized = self.position_size * (price - self.entry_price) if self.position_size > 0 else 0
@ -111,6 +137,42 @@ class BacktestEngine:
self.print_results()
def open_position(self, price, time):
qty_usd = self.pos_size_margin * self.leverage
qty_btc = qty_usd / price
fee = qty_usd * self.fee_rate
self.balance -= fee
if self.direction == "long":
self.position_size += qty_btc
else: # Short
self.position_size -= qty_btc
self.entry_price = price # Simplified avg entry
self.trades.append({"time": time, "type": f"Enter {self.direction.upper()}", "price": price, "fee": fee})
def close_partial_position(self, price, time):
qty_btc_exit = abs(self.position_size) * self.partial_exit_pct
self._close_qty(qty_btc_exit, price, time, "Partial Exit")
def close_full_position(self, price, time, reason="Exit"):
self._close_qty(abs(self.position_size), price, time, reason)
def _close_qty(self, qty_btc_exit, price, time, reason):
qty_usd_exit = qty_btc_exit * price
fee = qty_usd_exit * self.fee_rate
if self.direction == "long":
pnl = qty_btc_exit * (price - self.entry_price)
self.position_size -= qty_btc_exit
else: # Short
pnl = qty_btc_exit * (self.entry_price - price)
self.position_size += qty_btc_exit
self.balance += (pnl - fee)
self.trades.append({"time": time, "type": reason, "price": price, "pnl": pnl, "fee": fee})
def print_results(self):
total_pnl = self.equity - 1000.0
roi = (total_pnl / 1000.0) * 100
@ -127,16 +189,37 @@ class BacktestEngine:
print("="*30)
async def main():
engine = BacktestEngine()
# Assume BTC/1m for now
symbol = engine.config['symbol'].replace("USDT", "").replace("USD", "")
interval = engine.config['interval'] + "m"
parser = argparse.ArgumentParser(description='Ping-Pong Strategy Backtester')
parser.add_argument('--config', type=str, default='config/ping_pong_config.yaml', help='Path to config file')
parser.add_argument('--limit', type=int, default=10000, help='Max 1m candles (only if no dates)')
parser.add_argument('--start_date', type=str, help='Start Date (YYYY-MM-DD)')
parser.add_argument('--end_date', type=str, help='End Date (YYYY-MM-DD)')
parser.add_argument('--ma_period', type=int, help='MA Period for regime switching')
parser.add_argument('--ma_interval', type=str, default='1h', help='MA Interval (15m, 1h, 4h, 1d)')
parser.add_argument('--direction', type=str, choices=['long', 'short'], help='Fixed direction override')
df = await engine.load_data(symbol, interval, limit=10000)
if not df.empty:
engine.run(df)
else:
print("No data found in DB.")
args = parser.parse_args()
engine = BacktestEngine(config_path=args.config)
# Base Data (1m)
symbol = engine.config['symbol'].replace("USDT", "").replace("USD", "")
df = await engine.load_data(symbol, "1m", limit=args.limit, start_date=args.start_date, end_date=args.end_date)
if df.empty:
print("No 1m data found for this period.")
return
# MA Data (if enabled)
ma_df = None
if args.ma_period:
# Load slightly more MA candles before the start_date to initialize the MA correctly
ma_df = await engine.load_data(symbol, args.ma_interval, limit=5000, start_date=None, end_date=args.end_date)
if args.direction:
engine.direction = args.direction
engine.strategy.direction = args.direction
engine.run(df, ma_df=ma_df, ma_period=args.ma_period)
if __name__ == "__main__":
asyncio.run(main())