summaryrefslogtreecommitdiff
path: root/scripts
diff options
context:
space:
mode:
authorTheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>2026-04-02 10:26:52 +0900
committerTheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>2026-04-02 10:26:52 +0900
commit53cadcf7e34f05f77082e84f0696b56bcbcbae36 (patch)
treee02650e10c4d5727bc1e32e27788c17327fa46f7 /scripts
parentf5521da2876a2c19afc24f370b3258f2be95f81a (diff)
refactor: remove all crypto/Binance code, update to US stock symbols
Diffstat (limited to 'scripts')
-rwxr-xr-xscripts/optimize_asian_rsi.py234
1 files changed, 0 insertions, 234 deletions
diff --git a/scripts/optimize_asian_rsi.py b/scripts/optimize_asian_rsi.py
deleted file mode 100755
index 209453a..0000000
--- a/scripts/optimize_asian_rsi.py
+++ /dev/null
@@ -1,234 +0,0 @@
-#!/usr/bin/env python3
-"""Optimize Asian Session RSI strategy parameters via grid search.
-
-Uses synthetic SOL/USDT 5m candle data with dip-bounce patterns
-to exercise the strategy's RSI entry/exit logic across different
-parameter combinations.
-
-Usage: python scripts/optimize_asian_rsi.py
-"""
-
-import sys
-from pathlib import Path
-from decimal import Decimal
-from datetime import datetime, timedelta, timezone
-from typing import Optional
-import random
-
-# Add paths
-ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(ROOT / "services" / "strategy-engine" / "src"))
-sys.path.insert(0, str(ROOT / "services" / "strategy-engine"))
-sys.path.insert(0, str(ROOT / "services" / "backtester" / "src"))
-sys.path.insert(0, str(ROOT / "shared" / "src"))
-
-from shared.models import Candle, Signal # noqa: E402
-from backtester.engine import BacktestEngine # noqa: E402
-from strategies.asian_session_rsi import AsianSessionRsiStrategy # noqa: E402
-
-
-class OptimizableAsianRsi(AsianSessionRsiStrategy):
- """Subclass that bypasses ADX/EMA base-class filters for synthetic data.
-
- The base-class ``_apply_filters`` enforces ADX regime and EMA-level
- checks that are nearly impossible to satisfy on synthetic data (RSI
- oversold and price-above-EMA are mutually exclusive on random walks).
- This subclass passes signals through unchanged so that the core RSI
- entry/exit logic can be optimized independently.
- """
-
- def _apply_filters(self, signal: Optional[Signal]) -> Optional[Signal]:
- return signal
-
- def _price_above_ema(self) -> bool:
- # Bypass EMA filter for synthetic data optimisation
- return True
-
-
-def generate_sol_candles(days: int = 60, base_price: float = 150.0) -> list[Candle]:
- """Generate synthetic SOL/USDT 5-minute candles with dip-bounce patterns.
-
- Creates range-bound price action around *base_price* with periodic
- sharp dips during the Asian session window (00:00-02:00 UTC) followed
- by recovery bounces. This exercises the RSI oversold entry logic.
- """
- random.seed(42)
- candles: list[Candle] = []
- price = base_price
- start = datetime(2025, 1, 1, tzinfo=timezone.utc)
-
- for day in range(days):
- daily_trend = random.uniform(-0.003, 0.008)
-
- # ~50 % of days feature a V-dip during the Asian session
- dip_day = random.random() < 0.50
- dip_bar = random.randint(6, 18) if dip_day else -1
- dip_pct = random.uniform(0.015, 0.035)
-
- for bar in range(288): # 288 five-minute bars per day
- dt = start + timedelta(days=day, minutes=bar * 5)
- hour = dt.hour
-
- # Session-dependent volatility
- if 0 <= hour < 2:
- vol = 0.003
- elif 13 <= hour < 16:
- vol = 0.0025
- else:
- vol = 0.0015
-
- change = random.gauss(daily_trend / 288, vol)
- mean_rev = (base_price - price) / base_price * 0.001
- change += mean_rev
-
- # Inject V-dip: sharp single-bar crash then recovery
- session_bar = bar
- if dip_day and 0 <= hour < 2:
- if session_bar == dip_bar:
- change = -dip_pct
- elif session_bar == dip_bar + 1:
- change = dip_pct * random.uniform(0.5, 0.8)
- elif session_bar == dip_bar + 2:
- change = dip_pct * random.uniform(0.1, 0.3)
-
- open_p = price
- close_p = price * (1 + change)
- high_p = max(open_p, close_p) * (1 + abs(random.gauss(0, vol * 0.5)))
- low_p = min(open_p, close_p) * (1 - abs(random.gauss(0, vol * 0.5)))
-
- volume = random.uniform(50, 200)
- if 0 <= hour < 2:
- volume *= 2
- if dip_day and dip_bar <= session_bar <= dip_bar + 2:
- volume *= 2.5
-
- candles.append(
- Candle(
- symbol="SOLUSDT",
- timeframe="5m",
- open_time=dt,
- open=Decimal(str(round(open_p, 4))),
- high=Decimal(str(round(high_p, 4))),
- low=Decimal(str(round(low_p, 4))),
- close=Decimal(str(round(close_p, 4))),
- volume=Decimal(str(round(volume, 2))),
- )
- )
-
- price = close_p
-
- return candles
-
-
-def run_backtest(
- candles: list[Candle],
- params: dict,
- balance: float = 750.0,
- slippage: float = 0.001,
- fee: float = 0.001,
-):
- """Run a single backtest with given parameters."""
- strategy = OptimizableAsianRsi()
- strategy.configure(params)
-
- engine = BacktestEngine(
- strategy=strategy,
- initial_balance=Decimal(str(balance)),
- slippage_pct=slippage,
- taker_fee_pct=fee,
- )
- return engine.run(candles)
-
-
-def main() -> None:
- print("=" * 60)
- print("Asian Session RSI -- Parameter Optimization")
- print("SOL/USDT 5m | Capital: $750 (~100만원)")
- print("=" * 60)
-
- days = 60
- print(f"\nGenerating {days} days of synthetic SOL/USDT 5m candles...")
- candles = generate_sol_candles(days=days, base_price=150.0)
- print(f"Generated {len(candles)} candles")
- print("(base-class ADX/EMA filters bypassed for synthetic data)")
-
- # Parameter grid
- param_grid: list[dict] = []
- for rsi_period in [7, 9, 14]:
- for rsi_oversold in [20, 25, 30]:
- for tp in [1.0, 1.5, 2.0]:
- for sl in [0.5, 0.7, 1.0]:
- param_grid.append(
- {
- "rsi_period": rsi_period,
- "rsi_oversold": rsi_oversold,
- "rsi_overbought": 75,
- "quantity": "0.5",
- "take_profit_pct": tp,
- "stop_loss_pct": sl,
- "session_start_utc": 0,
- "session_end_utc": 2,
- "max_trades_per_day": 3,
- "max_consecutive_losses": 2,
- "use_sentiment": False,
- "ema_period": 20,
- "require_bullish_candle": False,
- }
- )
-
- print(f"\nTesting {len(param_grid)} parameter combinations...")
- print("-" * 60)
-
- results: list[tuple] = []
- for i, params in enumerate(param_grid):
- result = run_backtest(candles, params)
- sharpe = result.detailed.sharpe_ratio if result.detailed else 0.0
- results.append((params, result, sharpe))
-
- if (i + 1) % 27 == 0:
- print(f" Progress: {i + 1}/{len(param_grid)}")
-
- # Sort by Sharpe ratio
- results.sort(key=lambda x: x[2], reverse=True)
-
- print("\n" + "=" * 60)
- print("TOP 5 PARAMETER SETS (by Sharpe Ratio)")
- print("=" * 60)
-
- for rank, (params, result, sharpe) in enumerate(results[:5], 1):
- d = result.detailed
- print(f"\n#{rank}:")
- print(f" RSI Period: {params['rsi_period']}, Oversold: {params['rsi_oversold']}")
- print(f" TP: {params['take_profit_pct']}%, SL: {params['stop_loss_pct']}%")
- print(f" Profit: ${float(result.profit):.2f} ({float(result.profit_pct):.2f}%)")
- print(f" Trades: {result.total_trades}, Win Rate: {result.win_rate:.1f}%")
- if d:
- print(f" Sharpe: {d.sharpe_ratio:.3f}, Max DD: {d.max_drawdown:.2f}%")
- print(f" Profit Factor: {d.profit_factor:.2f}")
-
- # Also show worst 3 for comparison
- print("\n" + "=" * 60)
- print("WORST 3 PARAMETER SETS")
- print("=" * 60)
- for _rank, (params, result, sharpe) in enumerate(results[-3:], 1):
- print(
- f"\n RSI({params['rsi_period']}), OS={params['rsi_oversold']},"
- f" TP={params['take_profit_pct']}%, SL={params['stop_loss_pct']}%"
- )
- print(f" Profit: ${float(result.profit):.2f}, Trades: {result.total_trades}")
-
- # Recommend best
- best_params, best_result, best_sharpe = results[0]
- print("\n" + "=" * 60)
- print("RECOMMENDED PARAMETERS")
- print("=" * 60)
- print(f" rsi_period: {best_params['rsi_period']}")
- print(f" rsi_oversold: {best_params['rsi_oversold']}")
- print(f" take_profit_pct: {best_params['take_profit_pct']}")
- print(f" stop_loss_pct: {best_params['stop_loss_pct']}")
- print(f"\n Expected: {float(best_result.profit_pct):.2f}% over {days} days")
- print(f" Sharpe: {best_sharpe:.3f}")
-
-
-if __name__ == "__main__":
- main()