{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "8822ff95", "metadata": { "vscode": { "languageId": "plaintext" } }, "outputs": [], "source": [ "import MetaTrader5 as mt5\n", "import pandas as pd\n", "import numpy as np\n", "from sklearn.preprocessing import StandardScaler\n", "from sklearn.metrics import mean_absolute_error, mean_squared_error\n", "from datetime import datetime, timedelta\n", "\n", "# ----------------------------\n", "# PARAMETERS\n", "# ----------------------------\n", "SYMBOL = \"BTCUSDc\"\n", "TIMEFRAME = mt5.TIMEFRAME_D1\n", "LOOKBACK_DAYS = 365 * 7 # 7 years\n", "SEQ_LENGTH = 60\n", "FEATURES = ['Close', 'pctB', 'RSI', 'MACD', 'Signal_Line', 'Momentum', 'ATR']\n", "\n", "# ----------------------------\n", "# HELPER FUNCTIONS\n", "# ----------------------------\n", "def initialize_mt5():\n", " if not mt5.initialize():\n", " raise RuntimeError(f\"Failed to initialize MT5: {mt5.last_error()}\")\n", "\n", "def fetch_data(symbol, timeframe, lookback_days):\n", " utc_now = datetime.utcnow()\n", " from_date = utc_now - timedelta(days=lookback_days)\n", " rates = mt5.copy_rates_range(symbol, timeframe, from_date, utc_now)\n", " if rates is None or len(rates) == 0:\n", " raise RuntimeError(f\"No data retrieved for {symbol}\")\n", " df = pd.DataFrame(rates)\n", " df['time'] = pd.to_datetime(df['time'], unit='s')\n", " df.set_index('time', inplace=True)\n", " return df\n", "\n", "def compute_indicators(df):\n", " df['MA20'] = df['close'].rolling(20).mean()\n", " df['MA50'] = df['close'].rolling(50).mean()\n", " df['STD'] = df['close'].rolling(20).std()\n", " df['Upper_Band'] = df['MA20'] + (df['STD'] * 2.5)\n", " df['Lower_Band'] = df['MA20'] - (df['STD'] * 2.5)\n", " df['pctB'] = (df['close'] - df['Lower_Band']) / (df['Upper_Band'] - df['Lower_Band'])\n", "\n", " delta = df['close'].diff()\n", " up = delta.clip(lower=0)\n", " down = -delta.clip(upper=0)\n", " roll_up = up.rolling(14).mean()\n", " roll_down = down.rolling(14).mean()\n", " RS = roll_up / roll_down\n", " df['RSI'] = 100.0 - (100.0 / (1.0 + RS))\n", "\n", " exp1 = df['close'].ewm(span=12, adjust=False).mean()\n", " exp2 = df['close'].ewm(span=26, adjust=False).mean()\n", " df['MACD'] = exp1 - exp2\n", " df['Signal_Line'] = df['MACD'].ewm(span=9, adjust=False).mean()\n", "\n", " df['Momentum'] = df['close'] - df['close'].shift(10)\n", " df['TR'] = df[['high','close']].max(axis=1) - df[['low','close']].min(axis=1)\n", " df['ATR'] = df['TR'].rolling(14).mean()\n", "\n", " df.dropna(inplace=True)\n", " return df\n", "\n", "def generate_signals(df):\n", " # Simple backtest logic: Buy/Sell signals based on predicted vs RSI thresholds\n", " df['Predicted_Change'] = df['close'].pct_change().shift(-1) # naive predictor: next close % change\n", " rsi_buy = df['RSI'].quantile(0.4)\n", " rsi_sell = df['RSI'].quantile(0.6)\n", " pred_buy = df['Predicted_Change'].quantile(0.6)\n", " pred_sell = df['Predicted_Change'].quantile(0.4)\n", " df['Signal'] = 0\n", " df.loc[(df['Predicted_Change'] > pred_buy) & (df['RSI'] < rsi_buy), 'Signal'] = 1\n", " df.loc[(df['Predicted_Change'] < pred_sell) & (df['RSI'] > rsi_sell), 'Signal'] = -1\n", " return df\n", "\n", "def backtest(df, initial_capital=500.0, transaction_cost=0.0005, stop_loss_pct=0.1, take_profit_pct=0.2):\n", " cash = initial_capital\n", " holdings = 0\n", " entry_price = None\n", " positions = []\n", " portfolio_value = []\n", "\n", " for idx, row in df.iterrows():\n", " price = row['close']\n", " signal = row['Signal']\n", "\n", " # Enter long\n", " if signal == 1 and cash > 0:\n", " amount = cash * 0.5 * (1 - transaction_cost)\n", " holdings += amount / price\n", " cash -= amount\n", " entry_price = price\n", " positions.append({'Date': str(idx), 'Position': 'Buy', 'Price': price})\n", "\n", " # Exit long\n", " elif signal == -1 and holdings > 0:\n", " cash += holdings * price * (1 - transaction_cost)\n", " holdings = 0\n", " entry_price = None\n", " positions.append({'Date': str(idx), 'Position': 'Sell', 'Price': price})\n", "\n", " # Stop loss / take profit\n", " elif holdings > 0 and entry_price:\n", " if price <= entry_price * (1 - stop_loss_pct):\n", " cash += holdings * price * (1 - transaction_cost)\n", " holdings = 0\n", " entry_price = None\n", " positions.append({'Date': str(idx), 'Position': 'Stop Loss', 'Price': price})\n", " elif price >= entry_price * (1 + take_profit_pct):\n", " cash += holdings * price * (1 - transaction_cost)\n", " holdings = 0\n", " entry_price = None\n", " positions.append({'Date': str(idx), 'Position': 'Take Profit', 'Price': price})\n", "\n", " total_val = cash + holdings * price\n", " portfolio_value.append(total_val)\n", "\n", " df['Portfolio_Value'] = portfolio_value\n", " df['Daily_Return'] = df['Portfolio_Value'].pct_change()\n", " df['Cumulative_Return'] = (1 + df['Daily_Return'].fillna(0)).cumprod()\n", " total_return = (df['Portfolio_Value'].iloc[-1] - initial_capital) / initial_capital * 100\n", " return df, positions, total_return\n", "\n", "# ----------------------------\n", "# MAIN EXECUTION\n", "# ----------------------------\n", "if __name__ == \"__main__\":\n", " initialize_mt5()\n", " df = fetch_data(SYMBOL, TIMEFRAME, LOOKBACK_DAYS)\n", " df = compute_indicators(df)\n", " df = generate_signals(df)\n", " df, positions, total_return = backtest(df)\n", "\n", " print(\"Backtest complete!\")\n", " print(f\"Total return: {total_return:.2f}%\")\n", " print(\"Positions executed:\")\n", " for pos in positions[-5:]: # last 5 positions\n", " print(pos)\n", "\n", " mt5.shutdown()\n" ] } ], "metadata": { "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }