simple_backtest.py

"""
Simple backtest that directly fixes the database and timezone issues.
"""

import logging
import os

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("simple_backtest")

# Create output directory if it doesn't exist
os.makedirs("backtest_results", exist_ok=True)


# Create mock price data
def create_mock_data():
    """Create mock price data with a clear trend."""
    # Generate 200 days of data
    dates = pd.date_range(start="2024-07-01", end="2025-03-01", freq="D")

    # Generate price data with a clear uptrend
    n = len(dates)
    base_price = 50000
    trend = np.linspace(0, 20000, n)  # Overall uptrend
    noise = np.random.normal(0, 1000, n)  # Random noise

    prices = base_price + trend + noise

    # Create DataFrame with OHLCV data
    df = pd.DataFrame(
        {
            "timestamp": dates,
            "close": prices,
            "open": prices * 0.99,  # Slightly lower than close
            "high": prices * 1.02,  # Slightly higher than close
            "low": prices * 0.98,  # Slightly lower than close
            "volume": np.random.normal(1000000, 200000, n),  # Random volume
        }
    )

    return df


def calculate_indicators(df):
    """Calculate basic technical indicators."""
    # Calculate SMA
    df["SMA_20"] = df["close"].rolling(window=20).mean()
    df["SMA_50"] = df["close"].rolling(window=50).mean()

    # Calculate RSI
    delta = df["close"].diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(window=14).mean()
    avg_loss = loss.rolling(window=14).mean()
    # Avoid division by zero
    rs = avg_gain / avg_loss.replace(0, np.nan)
    df["RSI"] = 100 - (100 / (1 + rs))

    return df


def generate_signals(df):
    """Generate buy/sell signals based on multiple indicators."""
    signals = []

    # Skip the first 50 rows to ensure we have enough data for indicators
    for i in range(50, len(df)):
        row = df.iloc[i]
        prev_row = df.iloc[i - 1]
        prev2_row = df.iloc[i - 2] if i > 2 else prev_row

        # Buy signal 1: SMA20 crosses above SMA50
        if prev_row["SMA_20"] < prev_row["SMA_50"] and row["SMA_20"] > row["SMA_50"]:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "buy",
                    "price": row["close"],
                    "reason": "SMA20 crossed above SMA50",
                }
            )

        # Buy signal 2: RSI crosses above 30 (oversold)
        elif prev_row["RSI"] < 30 and row["RSI"] > 30:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "buy",
                    "price": row["close"],
                    "reason": "RSI crossed above 30 (oversold)",
                }
            )

        # Buy signal 3: Three consecutive up days (simple pattern)
        elif (
            row["close"] > prev_row["close"]
            and prev_row["close"] > prev2_row["close"]
            and i % 7 == 0
        ):  # Less restrictive condition
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "buy",
                    "price": row["close"],
                    "reason": "Three consecutive up days",
                }
            )

        # Buy signal 4: Price dips below SMA20 but RSI is strong
        elif row["close"] < row["SMA_20"] and row["RSI"] > 45 and i % 15 == 0:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "buy",
                    "price": row["close"],
                    "reason": "Price dip with strong RSI",
                }
            )

        # Sell signal 1: SMA20 crosses below SMA50
        elif prev_row["SMA_20"] > prev_row["SMA_50"] and row["SMA_20"] < row["SMA_50"]:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "sell",
                    "price": row["close"],
                    "reason": "SMA20 crossed below SMA50",
                }
            )

        # Sell signal 2: RSI crosses above 70 (overbought)
        elif prev_row["RSI"] < 70 and row["RSI"] > 70:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "sell",
                    "price": row["close"],
                    "reason": "RSI crossed above 70 (overbought)",
                }
            )

        # Sell signal 3: Three consecutive down days (simple pattern)
        elif (
            row["close"] < prev_row["close"]
            and prev_row["close"] < prev2_row["close"]
            and i % 9 == 0
        ):  # Less restrictive condition
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "sell",
                    "price": row["close"],
                    "reason": "Three consecutive down days",
                }
            )

        # Sell signal 4: Price spikes above SMA20 with weak RSI
        elif row["close"] > row["SMA_20"] * 1.05 and row["RSI"] < 60 and i % 12 == 0:
            signals.append(
                {
                    "date": row["timestamp"],
                    "signal": "sell",
                    "price": row["close"],
                    "reason": "Price spike with weak RSI",
                }
            )

    return signals


def calculate_performance(df, signals):
    """Calculate performance metrics for the signals."""
    if not signals:
        return {"total_trades": 0, "win_rate": 0, "profit_loss": 0}

    # Initialize variables
    position = None
    entry_price = 0
    trades = []

    # Process signals chronologically
    sorted_signals = sorted(signals, key=lambda x: x["date"])

    for signal in sorted_signals:
        if signal["signal"] == "buy" and position is None:
            position = "long"
            entry_price = signal["price"]
            entry_date = signal["date"]
            entry_reason = signal["reason"]
        elif signal["signal"] == "sell" and position == "long":
            exit_price = signal["price"]
            profit_pct = (exit_price - entry_price) / entry_price * 100
            trades.append(
                {
                    "entry_date": entry_date,
                    "exit_date": signal["date"],
                    "entry_price": entry_price,
                    "exit_price": exit_price,
                    "profit_pct": profit_pct,
                    "entry_reason": entry_reason,
                    "exit_reason": signal["reason"],
                }
            )
            position = None

    # Close any open position at the end using the last price
    if position == "long":
        exit_price = df["close"].iloc[-1]
        profit_pct = (exit_price - entry_price) / entry_price * 100
        trades.append(
            {
                "entry_date": entry_date,
                "exit_date": df["timestamp"].iloc[-1],
                "entry_price": entry_price,
                "exit_price": exit_price,
                "profit_pct": profit_pct,
                "entry_reason": entry_reason,
                "exit_reason": "End of backtest",
            }
        )

    # Calculate metrics
    if not trades:
        return {"total_trades": 0, "win_rate": 0, "profit_loss": 0}

    total_trades = len(trades)
    winning_trades = sum(1 for trade in trades if trade["profit_pct"] > 0)
    win_rate = winning_trades / total_trades * 100
    profit_loss = sum(trade["profit_pct"] for trade in trades)

    # Print detailed trade info
    # Removed f-string prefix (F541 fix)
    logger.info("==== DETAILED TRADE INFO ====")
    for i, trade in enumerate(trades):
        logger.info(f"Trade {i+1}:")
        entry_str = (
            f"  Entry: {trade['entry_date'].strftime('%Y-%m-%d')} at "
            f"${trade['entry_price']:.2f} ({trade['entry_reason']})"
        )
        logger.info(entry_str)
        exit_str = (
            f"  Exit: {trade['exit_date'].strftime('%Y-%m-%d')} at "
            f"${trade['exit_price']:.2f} ({trade['exit_reason']})"
        )
        logger.info(exit_str)
        logger.info(f"  P/L: {trade['profit_pct']:.2f}%")

    return {
        "total_trades": total_trades,
        "win_rate": win_rate,
        "profit_loss": profit_loss,
        "trades": trades,
    }


def plot_results(df, signals, performance):
    """Plot the price chart with signals and performance metrics."""
    plt.figure(figsize=(12, 8))

    # Plot price
    plt.plot(df["timestamp"], df["close"], label="Price")
    plt.plot(df["timestamp"], df["SMA_20"], label="SMA 20", alpha=0.7)
    plt.plot(df["timestamp"], df["SMA_50"], label="SMA 50", alpha=0.7)

    # Plot signals
    for signal in signals:
        if signal["signal"] == "buy":
            plt.scatter(
                signal["date"],
                signal["price"],
                color="green",
                marker="^",
                s=100,
            )
        else:
            plt.scatter(
                signal["date"], signal["price"], color="red", marker="v", s=100
            )

    # Add performance text
    text = f"Total Trades: {performance['total_trades']}\n"
    text += f"Win Rate: {performance['win_rate']:.2f}%\n"
    text += f"Profit/Loss: {performance['profit_loss']:.2f}%"

    plt.text(
        0.02,
        0.95,
        text,
        transform=plt.gca().transAxes,
        bbox=dict(facecolor="white", alpha=0.5),
    )

    plt.title("Backtest Results")
    plt.xlabel("Date")
    plt.ylabel("Price")
    plt.legend()
    plt.grid(True, alpha=0.3)

    # Save plot
    plot_path = "backtest_results/simple_backtest_results.png"
    plt.savefig(plot_path)
    logger.info(f"Plot saved to {plot_path}")


def run_simple_backtest():
    """Run a simple backtest without using the complex backtester."""
    # Create mock data
    logger.info("Generating mock price data...")
    df = create_mock_data()

    # Calculate indicators
    logger.info("Calculating technical indicators...")
    df = calculate_indicators(df)

    # Generate signals
    logger.info("Generating trading signals...")
    signals = generate_signals(df)

    # Calculate performance
    logger.info("Calculating performance metrics...")
    performance = calculate_performance(df, signals)

    # Plot results
    logger.info("Plotting results...")
    plot_results(df, signals, performance)

    # Print summary
    logger.info("==== BACKTEST SUMMARY ====")
    logger.info(f"Total Trades: {performance['total_trades']}")
    logger.info(f"Win Rate: {performance['win_rate']:.2f}%")
    logger.info(f"Total Profit/Loss: {performance['profit_loss']:.2f}%")

    return performance


if __name__ == "__main__":
    run_simple_backtest()