diff options
Diffstat (limited to 'scripts')
| -rwxr-xr-x | scripts/backtest_moc.py | 246 | ||||
| -rwxr-xr-x | scripts/optimize_asian_rsi.py | 214 | ||||
| -rwxr-xr-x | scripts/stock_screener.py | 72 |
3 files changed, 298 insertions, 234 deletions
diff --git a/scripts/backtest_moc.py b/scripts/backtest_moc.py new file mode 100755 index 0000000..92b426b --- /dev/null +++ b/scripts/backtest_moc.py @@ -0,0 +1,246 @@ +#!/usr/bin/env python3 +"""Backtest and optimize MOC strategy on synthetic US stock data. + +Usage: python scripts/backtest_moc.py +""" + +import sys +import random +from pathlib import Path +from decimal import Decimal +from datetime import datetime, timedelta, timezone + +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 # noqa: E402 +from backtester.engine import BacktestEngine # noqa: E402 +from strategies.moc_strategy import MocStrategy # noqa: E402 + + +def generate_stock_candles( + symbol: str = "AAPL", + days: int = 90, + base_price: float = 180.0, + daily_drift: float = 0.0003, # Slight upward bias + daily_vol: float = 0.015, # 1.5% daily vol +) -> list[Candle]: + """Generate realistic US stock intraday candles (5-min bars). + + Simulates: + - Market hours only (14:30-21:00 UTC = 9:30-16:00 ET) + - Opening gaps (overnight news effect) + - Intraday volatility pattern (higher at open/close) + - Volume pattern (U-shaped: high at open, low midday, high at close) + """ + candles = [] + price = base_price + start_date = datetime(2025, 1, 2, tzinfo=timezone.utc) # Start on a Thursday + + trading_day = 0 + current_date = start_date + + while trading_day < days: + # Skip weekends + if current_date.weekday() >= 5: + current_date += timedelta(days=1) + continue + + # Opening gap: overnight news effect + gap_pct = random.gauss(daily_drift, daily_vol * 0.5) # Gap is ~50% of daily vol + price *= 1 + gap_pct + + # Generate 78 5-minute bars (9:30-16:00 = 6.5 hours = 78 bars) + intraday_bars = 78 + for bar in range(intraday_bars): + # Time: 14:30 UTC + bar * 5 minutes + dt = current_date.replace(hour=14, minute=30) + timedelta(minutes=bar * 5) + + # Intraday volatility pattern (U-shaped) + hour_of_day = bar / intraday_bars + if hour_of_day < 0.1: # First 10% of day (opening) + vol = daily_vol * 0.003 + elif hour_of_day > 0.9: # Last 10% (closing) + vol = daily_vol * 0.0025 + else: # Middle of day + vol = daily_vol * 0.001 + + # Add daily trend component + intraday_drift = daily_drift / intraday_bars + change = random.gauss(intraday_drift, vol) + + open_p = price + close_p = price * (1 + change) + high_p = max(open_p, close_p) * (1 + abs(random.gauss(0, vol * 0.3))) + low_p = min(open_p, close_p) * (1 - abs(random.gauss(0, vol * 0.3))) + + # Volume pattern (U-shaped) + if hour_of_day < 0.1 or hour_of_day > 0.85: + volume = random.uniform(500000, 2000000) + else: + volume = random.uniform(100000, 500000) + + candles.append( + Candle( + symbol=symbol, + timeframe="5Min", + open_time=dt, + open=Decimal(str(round(open_p, 2))), + high=Decimal(str(round(high_p, 2))), + low=Decimal(str(round(low_p, 2))), + close=Decimal(str(round(close_p, 2))), + volume=Decimal(str(int(volume))), + ) + ) + + price = close_p + + trading_day += 1 + current_date += timedelta(days=1) + + return candles + + +def run_backtest(candles, params, balance=750.0): + """Run a single backtest.""" + strategy = MocStrategy() + strategy.configure(params) + + engine = BacktestEngine( + strategy=strategy, + initial_balance=Decimal(str(balance)), + slippage_pct=0.0005, # 0.05% slippage (stocks have tighter spreads) + taker_fee_pct=0.0, # Alpaca = 0% commission + ) + return engine.run(candles) + + +def main(): + random.seed(42) + + print("=" * 60) + print("MOC Strategy Backtest — US Stocks") + print("Capital: $750 (~100만원)") + print("=" * 60) + + # Test across multiple stocks + stocks = [ + ("AAPL", 180.0, 0.0003, 0.015), + ("MSFT", 420.0, 0.0004, 0.014), + ("TSLA", 250.0, 0.0001, 0.030), + ("NVDA", 800.0, 0.0005, 0.025), + ("AMZN", 185.0, 0.0003, 0.018), + ] + + # Parameter grid + param_sets = [] + for rsi_min in [25, 30, 35]: + for rsi_max in [55, 60, 65]: + for sl in [1.5, 2.0, 3.0]: + for ema in [10, 20]: + param_sets.append( + { + "quantity_pct": 0.2, + "stop_loss_pct": sl, + "rsi_min": rsi_min, + "rsi_max": rsi_max, + "ema_period": ema, + "volume_avg_period": 20, + "min_volume_ratio": 0.8, + "buy_start_utc": 19, + "buy_end_utc": 21, + "sell_start_utc": 14, + "sell_end_utc": 15, + "max_positions": 5, + } + ) + + print(f"\nParameter combinations: {len(param_sets)}") + print(f"Stocks: {[s[0] for s in stocks]}") + print("Generating 90 days of 5-min data per stock...\n") + + # Generate data for each stock + all_candles = {} + for symbol, base, drift, vol in stocks: + all_candles[symbol] = generate_stock_candles( + symbol, days=90, base_price=base, daily_drift=drift, daily_vol=vol + ) + print(f" {symbol}: {len(all_candles[symbol])} candles") + + # Test each parameter set across all stocks + print( + f"\nRunning {len(param_sets)} x {len(stocks)} = {len(param_sets) * len(stocks)} backtests..." + ) + + param_results = [] + for i, params in enumerate(param_sets): + total_profit = Decimal("0") + total_trades = 0 + total_sharpe = 0.0 + stock_details = [] + + for symbol, _, _, _ in stocks: + result = run_backtest(all_candles[symbol], params) + total_profit += result.profit + total_trades += result.total_trades + if result.detailed: + total_sharpe += result.detailed.sharpe_ratio + stock_details.append((symbol, result)) + + avg_sharpe = total_sharpe / len(stocks) if stocks else 0 + param_results.append((params, total_profit, total_trades, avg_sharpe, stock_details)) + + if (i + 1) % 18 == 0: + print(f" Progress: {i + 1}/{len(param_sets)}") + + # Sort by average Sharpe + param_results.sort(key=lambda x: x[3], reverse=True) + + print("\n" + "=" * 60) + print("TOP 5 PARAMETER SETS (by avg Sharpe across all stocks)") + print("=" * 60) + + for rank, (params, profit, trades, sharpe, details) in enumerate(param_results[:5], 1): + print(f"\n#{rank}:") + print( + f" RSI: {params['rsi_min']}-{params['rsi_max']}," + f" SL: {params['stop_loss_pct']}%, EMA: {params['ema_period']}" + ) + print(f" Total Profit: ${float(profit):.2f}, Trades: {trades}, Avg Sharpe: {sharpe:.3f}") + print(" Per stock:") + for symbol, result in details: + pct = float(result.profit_pct) + dd = result.detailed.max_drawdown if result.detailed else 0 + print(f" {symbol}: {pct:+.2f}% ({result.total_trades} trades, DD: {dd:.1f}%)") + + # Best params + best = param_results[0] + print("\n" + "=" * 60) + print("RECOMMENDED PARAMETERS") + print("=" * 60) + bp = best[0] + print(f" rsi_min: {bp['rsi_min']}") + print(f" rsi_max: {bp['rsi_max']}") + print(f" stop_loss_pct: {bp['stop_loss_pct']}") + print(f" ema_period: {bp['ema_period']}") + print(f" min_volume_ratio: {bp['min_volume_ratio']}") + print(f"\n Avg Sharpe: {best[3]:.3f}") + print(f" Total Profit: ${float(best[1]):.2f} across 5 stocks over 90 days") + + # Worst for comparison + print("\n" + "=" * 60) + print("WORST 3 PARAMETER SETS") + print("=" * 60) + for _rank, (params, profit, trades, sharpe, _) in enumerate(param_results[-3:], 1): + print( + f" RSI({params['rsi_min']}-{params['rsi_max']})," + f" SL={params['stop_loss_pct']}%, EMA={params['ema_period']}" + ) + print(f" Profit: ${float(profit):.2f}, Sharpe: {sharpe:.3f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/optimize_asian_rsi.py b/scripts/optimize_asian_rsi.py deleted file mode 100755 index 9921447..0000000 --- a/scripts/optimize_asian_rsi.py +++ /dev/null @@ -1,214 +0,0 @@ -#!/usr/bin/env python3 -"""Optimize Asian Session RSI strategy parameters via grid search. - -Usage: python scripts/optimize_asian_rsi.py -""" - -import sys -from pathlib import Path -from decimal import Decimal -from datetime import datetime, timedelta, timezone -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 # noqa: E402 -from backtester.engine import BacktestEngine # noqa: E402 -from strategies.asian_session_rsi import AsianSessionRsiStrategy # noqa: E402 - - -def generate_sol_candles(days: int = 90, base_price: float = 150.0) -> list[Candle]: - """Generate realistic SOL/USDT 5-minute candles. - - Simulates: - - Mild uptrend with periodic sharp dips during Asian session - - Intraday volatility (higher at session opens) - - Random walk with mean reversion - - Occasional momentum bursts that create RSI extremes - """ - random.seed(42) - candles = [] - price = base_price - start = datetime(2025, 1, 1, tzinfo=timezone.utc) - - for day in range(days): - # Mild upward bias to keep price above EMA - daily_trend = random.uniform(-0.005, 0.015) - - # Many days have a sharp V-dip during Asian session (1-2 bar crash + recovery) - # This creates RSI oversold while EMA stays above price briefly - dip_day = random.random() < 0.45 - dip_bar = random.randint(4, 18) if dip_day else -1 - # Sharp single-bar dip: 2-4% drop then immediate recovery - dip_magnitude = random.uniform(0.02, 0.04) - - for bar in range(288): # 288 5-minute bars per day - dt = start + timedelta(days=day, minutes=bar * 5) - hour = dt.hour - - # Volatility varies by session - if 0 <= hour < 2: # Asian open (our trading window) - vol = 0.003 - elif 13 <= hour < 16: # US session - vol = 0.0025 - else: - vol = 0.0015 - - # Base random walk with upward drift - change = random.gauss(daily_trend / 288, vol) - mean_rev = (base_price - price) / base_price * 0.001 - change += mean_rev - - # Session bar index within 00:00-01:55 UTC (bars 0-23) - session_bar = bar - - # Inject sharp V-dip: 1 bar crash, 1 bar partial recovery - if dip_day and 0 <= hour < 2: - if session_bar == dip_bar: - # Crash bar: sharp drop - change = -dip_magnitude - elif session_bar == dip_bar + 1: - # Recovery bar: bounce back most of the way - change = dip_magnitude * random.uniform(0.5, 0.8) - elif session_bar == dip_bar + 2: - # Continued recovery - change = dip_magnitude * 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 # Spike volume on dip/recovery - - 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, params, balance=750.0, slippage=0.001, fee=0.001): - """Run a single backtest with given parameters.""" - strategy = AsianSessionRsiStrategy() - 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(): - print("=" * 60) - print("Asian Session RSI — Parameter Optimization") - print("SOL/USDT 5m | Capital: $750 (~100만원)") - print("=" * 60) - - days = 30 - 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") - - # Parameter grid - param_grid = [] - 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 = [] - 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']}, 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() diff --git a/scripts/stock_screener.py b/scripts/stock_screener.py index 387bfea..7a5c0ba 100755 --- a/scripts/stock_screener.py +++ b/scripts/stock_screener.py @@ -10,6 +10,7 @@ Usage: Requires: ANTHROPIC_API_KEY environment variable """ + import argparse import asyncio import json @@ -51,15 +52,17 @@ async def get_market_data(alpaca: AlpacaClient, symbols: list[str]) -> list[dict lows = [float(b["l"]) for b in bars] range_pct = (max(highs) - min(lows)) / close * 100 if close > 0 else 0 - results.append({ - "symbol": symbol, - "close": close, - "change_pct": round(change_pct, 2), - "volume": volume, - "vol_ratio": round(vol_ratio, 2), - "range_5d_pct": round(range_pct, 2), - "is_bullish": close > float(latest["o"]), # Today bullish? - }) + results.append( + { + "symbol": symbol, + "close": close, + "change_pct": round(change_pct, 2), + "volume": volume, + "vol_ratio": round(vol_ratio, 2), + "range_5d_pct": round(range_pct, 2), + "is_bullish": close > float(latest["o"]), # Today bullish? + } + ) except Exception as exc: print(f" Warning: Failed to fetch {symbol}: {exc}") @@ -137,19 +140,45 @@ async def analyze_with_claude(prompt: str, api_key: str) -> list[dict]: # Default universe of liquid US stocks DEFAULT_UNIVERSE = [ # Tech - "AAPL", "MSFT", "GOOGL", "AMZN", "META", "NVDA", "TSLA", "AMD", "INTC", "CRM", + "AAPL", + "MSFT", + "GOOGL", + "AMZN", + "META", + "NVDA", + "TSLA", + "AMD", + "INTC", + "CRM", # Finance - "JPM", "BAC", "GS", "MS", "V", "MA", + "JPM", + "BAC", + "GS", + "MS", + "V", + "MA", # Healthcare - "JNJ", "UNH", "PFE", "ABBV", + "JNJ", + "UNH", + "PFE", + "ABBV", # Consumer - "WMT", "KO", "PEP", "MCD", "NKE", + "WMT", + "KO", + "PEP", + "MCD", + "NKE", # Energy - "XOM", "CVX", + "XOM", + "CVX", # Industrial - "CAT", "BA", "GE", + "CAT", + "BA", + "GE", # ETFs - "SPY", "QQQ", "IWM", + "SPY", + "QQQ", + "IWM", ] @@ -187,8 +216,7 @@ async def main_async(top_n: int = 5, universe: list[str] | None = None): # Pre-filter obvious rejects candidates = [ - s for s in market_data - if s["is_bullish"] and s["vol_ratio"] >= 0.8 and s["change_pct"] > -2 + s for s in market_data if s["is_bullish"] and s["vol_ratio"] >= 0.8 and s["change_pct"] > -2 ] print(f"Pre-filter: {len(candidates)} candidates (bullish, decent volume)\n") @@ -211,7 +239,9 @@ async def main_async(top_n: int = 5, universe: list[str] | None = None): # Find market data for this symbol md = next((m for m in market_data if m["symbol"] == rec["symbol"]), None) if md: - print(f" Close: ${md['close']:.2f} | Change: {md['change_pct']:+.2f}% | Vol Ratio: {md['vol_ratio']:.1f}x") + print( + f" Close: ${md['close']:.2f} | Change: {md['change_pct']:+.2f}% | Vol Ratio: {md['vol_ratio']:.1f}x" + ) else: print("Claude found no qualifying stocks today.") else: @@ -237,7 +267,9 @@ async def main_async(top_n: int = 5, universe: list[str] | None = None): print("-" * 60) for i, (s, score) in enumerate(scored[:top_n], 1): print(f"#{i} {s['symbol']} (Score: {score})") - print(f" Close: ${s['close']:.2f} | Change: {s['change_pct']:+.2f}% | Vol: {s['vol_ratio']:.1f}x") + print( + f" Close: ${s['close']:.2f} | Change: {s['change_pct']:+.2f}% | Vol: {s['vol_ratio']:.1f}x" + ) print("\n" + "=" * 60) |