Ready Strategies

import pandas as pd
import numpy as np
from rlxbt import Strategy, rlx


class FalconATR(Strategy):
    def __init__(
        self,
        fast_ma: int = 5,
        slow_ma: int = 15,
        rsi_period: int = 14,
        rsi_oversold: float = 35,
        rsi_overbought: float = 65,
        atr_period: int = 14,
        tp_atr_multiplier: float = 2.0,  # TP = 2x ATR
        sl_atr_multiplier: float = 1.0,  # SL = 1x ATR
        momentum_period: int = 5,
        use_atr_based_levels: bool = True,
    ):
        super().__init__()
        self.fast_ma = fast_ma
        self.slow_ma = slow_ma
        self.rsi_period = rsi_period
        self.rsi_oversold = rsi_oversold
        self.rsi_overbought = rsi_overbought
        self.atr_period = atr_period
        self.tp_atr_multiplier = tp_atr_multiplier
        self.sl_atr_multiplier = sl_atr_multiplier
        self.momentum_period = momentum_period
        self.use_atr_based_levels = use_atr_based_levels

    def calculate_rsi(self, prices: pd.Series) -> pd.Series:
        delta = prices.diff()
        gain = delta.where(delta > 0, 0).rolling(window=self.rsi_period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))

    def calculate_atr(self, df: pd.DataFrame) -> pd.Series:
        high_low = df["high"] - df["low"]
        high_close = (df["high"] - df["close"].shift()).abs()
        low_close = (df["low"] - df["close"].shift()).abs()
        tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
        return tr.rolling(window=self.atr_period).mean()

    def calculate_momentum(self, prices: pd.Series) -> pd.Series:
        return prices.pct_change(self.momentum_period)

    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        df = data.copy()
        
        # Calculate indicators
        df["fast_ma"] = df["close"].rolling(self.fast_ma).mean()
        df["slow_ma"] = df["close"].rolling(self.slow_ma).mean()
        df["rsi"] = self.calculate_rsi(df["close"])
        df["atr"] = self.calculate_atr(df)
        df["momentum"] = self.calculate_momentum(df["close"])

        # Boolean conditions
        df["ma_bull"] = df["fast_ma"] > df["slow_ma"]
        df["ma_bear"] = df["fast_ma"] < df["slow_ma"]
        df["rsi_oversold"] = df["rsi"] < self.rsi_oversold
        df["rsi_overbought"] = df["rsi"] > self.rsi_overbought
        df["momentum_positive"] = df["momentum"] > 0.001
        df["momentum_negative"] = df["momentum"] < -0.001

        min_periods = max(self.slow_ma, self.rsi_period, self.atr_period)
        records = []

        for i in range(len(df)):
            if i < min_periods:
                records.append({"signal": 0, "take_profit": None, "stop_loss": None})
                continue

            row = df.iloc[i]
            price = row.close
            atr = row.atr

            # Buy signal: bearish MA cross + overbought RSI + negative momentum
            if row.ma_bear and row.rsi_overbought and row.momentum_negative:
                if self.use_atr_based_levels:
                    tp = price + atr * self.tp_atr_multiplier
                    sl = price - atr * self.sl_atr_multiplier
                else:
                    tp = price * 1.02
                    sl = price * 0.99
                records.append({"signal": 1, "take_profit": tp, "stop_loss": sl})
                continue

            # Sell signal: bullish MA cross + oversold RSI + positive momentum
            if row.ma_bull and row.rsi_oversold and row.momentum_positive:
                if self.use_atr_based_levels:
                    tp = price - atr * self.tp_atr_multiplier
                    sl = price + atr * self.sl_atr_multiplier
                else:
                    tp = price * 0.98
                    sl = price * 1.01
                records.append({"signal": -1, "take_profit": tp, "stop_loss": sl})
                continue

            records.append({"signal": 0, "take_profit": None, "stop_loss": None})

        result_df = pd.DataFrame.from_records(records)
        result_df.index = df.index
        return result_df

def run_falcon_strategy():
    # Load 1-minute data
    df_1m = pd.read_csv("data/BTCUSDT_1m.csv")
    df_1m["datetime"] = pd.to_datetime(df_1m["open_time"])
    df_1m = df_1m.set_index("datetime")

    # Resample to 9-minute bars for signals
    df_9m = df_1m.resample("9min").agg({
        "open": "first",
        "high": "max",
        "low": "min",
        "close": "last",
        "volume": "sum",
    }).dropna()

    # Create strategy with aggressive parameters
    strategy = FalconATR(
        fast_ma=3,
        slow_ma=7,
        rsi_period=7,
        rsi_oversold=40,
        rsi_overbought=60,
        atr_period=14,
        tp_atr_multiplier=0.3,  # Tight TP
        sl_atr_multiplier=2.0,  # Wide SL
    )

    # Use engine directly for intrabar backtest
    engine = rlx.TradingEngine(
        initial_capital=100_000.0,
        commission=0.001,
        slippage=0.0,
        contract_size=1.0,
        enable_dynamic_tp_sl=False,
    )

    # Run intrabar backtest (signals on 9m, execution on 1m)
    print("🚀 Running Falcon ATR Intrabar Backtest...")
    result = engine.run_intrabar_backtest(strategy, df_9m, df_1m)

    print(f"Total Return: {result.total_return * 100:.2f}%")
    print(f"Total Trades: {result.total_trades}")

    # Generate dashboard with intrabar resolution
    dashboard_gen = rlx.DashboardGenerator(
        initial_capital=100_000.0,
        commission=0.001,
        slippage=0.0,
        use_intrabar_resolution=True,  # Important!
        contract_size=1.0,
    )

    dashboard_result = dashboard_gen.generate_dashboard(result, df_9m, df_1m)
    dashboard_gen.plot(dashboard_result, port=8000, auto_open=True)