market/Cointegration Based Crypto Pairs Trading Strategy.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d6434fa8-c6d8-4305-8459-7937f98765ed",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "import time\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from datetime import datetime, timedelta\n",
    "from statsmodels.tsa.stattools import coint, adfuller\n",
    "import statsmodels.api as sm\n",
    "import matplotlib.pyplot as plt\n",
    "from statsmodels.regression.rolling import RollingOLS\n",
    "import matplotlib.dates as mdates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ef493366-0faf-4138-9bf4-2a83d749ee35",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Fetch Prices"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9c313824-9b49-42cf-8549-e4583e58e510",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def fetch_crypto_prices(pairs, exchanges, start_date, end_date):\n",
    "    \"\"\"Fetch cryptocurrency price data with pagination handling\"\"\"\n",
    "    start_dt = datetime.strptime(start_date, '%d-%m-%Y')\n",
    "    end_dt = datetime.strptime(end_date, '%d-%m-%Y')\n",
    "    \n",
    "    headers = {\n",
    "        \"accept\": \"application/json\",\n",
    "        \"Accept-Encoding\": \"gzip, deflate, br\",\n",
    "        \"x-api-key\": # Insert your actual API key here\n",
    "    }\n",
    "    \n",
    "    all_data = []\n",
    "    \n",
    "    for pair in pairs:\n",
    "        for exchange in exchanges:\n",
    "            print(f\"\\nFetching {pair} on {exchange}\")\n",
    "            pair_records = 0\n",
    "            current_start = start_dt\n",
    "            \n",
    "            while current_start < end_dt:\n",
    "                chunk_end = min(current_start + timedelta(days=730), end_dt)\n",
    "                print(f\"Period: {current_start.date()} to {chunk_end.date()}\")\n",
    "                \n",
    "                next_url = None\n",
    "                page_count = 1\n",
    "                \n",
    "                while True:\n",
    "                    try:\n",
    "                        if next_url:\n",
    "                            print(f\"Fetching page {page_count}...\", end=\" \")\n",
    "                            response = requests.get(next_url, headers=headers)\n",
    "                        else:\n",
    "                            url = f\"https://api.amberdata.com/markets/spot/ohlcv/{pair}\"\n",
    "                            params = {\n",
    "                                'exchange': exchange,\n",
    "                                'startDate': current_start.isoformat(),\n",
    "                                'endDate': chunk_end.isoformat(),\n",
    "                                'timeFormat': 'iso',\n",
    "                                'timeInterval': 'days'\n",
    "                            }\n",
    "                            response = requests.get(url, headers=headers, params=params)\n",
    "                        \n",
    "                        if response.status_code == 200:\n",
    "                            result = response.json()\n",
    "                            if 'payload' not in result or 'data' not in result['payload']:\n",
    "                                print(f\"No data available\")\n",
    "                                break\n",
    "                                \n",
    "                            data = result['payload']['data']\n",
    "                            if not data:  # Empty data array\n",
    "                                print(f\"No records found\")\n",
    "                                break\n",
    "                                \n",
    "                            records_in_page = len(data)\n",
    "                            pair_records += records_in_page\n",
    "                            all_data.extend(data)\n",
    "                            \n",
    "                            next_url = result['payload']['metadata'].get('next')\n",
    "                            if next_url:\n",
    "                                print(f\"Got {records_in_page} records\", end=\" \")\n",
    "                                page_count += 1\n",
    "                            else:\n",
    "                                print(f\"Got {records_in_page} records - Complete\")\n",
    "                                break\n",
    "                        else:\n",
    "                            error_msg = f\"Failed! Status code: {response.status_code}\"\n",
    "                            if response.status_code == 429:\n",
    "                                error_msg += \" (Rate limit exceeded - waiting 60 seconds)\"\n",
    "                                print(error_msg)\n",
    "                                time.sleep(60)\n",
    "                                continue\n",
    "                            else:\n",
    "                                print(error_msg)\n",
    "                                break\n",
    "                            \n",
    "                    except Exception as e:\n",
    "                        print(f\"Error: {str(e)}\")\n",
    "                        break\n",
    "                \n",
    "                current_start = chunk_end\n",
    "            \n",
    "            print(f\"Total records for {pair} on {exchange}: {pair_records}\")\n",
    "    \n",
    "    print(\"\\nProcessing downloaded data...\")\n",
    "    df = pd.DataFrame(all_data)\n",
    "    \n",
    "    if len(df) == 0:\n",
    "        print(\"No data was downloaded!\")\n",
    "        return pd.DataFrame()\n",
    "    \n",
    "    df['exchangeTimestamp'] = pd.to_datetime(df['exchangeTimestamp'])\n",
    "    df = df.sort_values('exchangeTimestamp').drop_duplicates(\n",
    "        subset=['instrument', 'exchange', 'exchangeTimestamp'], \n",
    "        keep='first'\n",
    "    )\n",
    "    \n",
    "    print(f\"\\nFinal dataset summary:\")\n",
    "    for instrument in df['instrument'].unique():\n",
    "        instrument_data = df[df['instrument'] == instrument]\n",
    "        date_range = instrument_data['exchangeTimestamp']\n",
    "        print(f\"{instrument:10} | {len(instrument_data):5} records | {date_range.min().date()} to {date_range.max().date()}\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1989873d-ba54-4f1b-869a-f54ba910a417",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Calculate Returns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dba8551b-66e6-4ad8-9ecc-8837bfc3d1a4",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def calculate_returns(df):\n",
    "    \"\"\"Calculate percentage returns from price data\"\"\"\n",
    "    return df['close'].pct_change()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6dc37b1f-5940-414c-a67e-28e7dc9765c7",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Test for Stationarity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "61104c4e-a0d0-4fc1-a591-8f30ac519e8d",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def test_stationarity(series, threshold=0.05):\n",
    "    \"\"\"\n",
    "    Test for stationarity using Augmented Dickey-Fuller test on the entire series.\n",
    "    \"\"\"\n",
    "    if not isinstance(series, pd.Series):\n",
    "        series = pd.Series(series)\n",
    "\n",
    "    try:\n",
    "        adf_test = adfuller(series.dropna(), regression='c', maxlag=5)\n",
    "        return {\n",
    "            'is_stationary': adf_test[1] < threshold,\n",
    "            'adf_statistic': adf_test[0],\n",
    "            'p_value': adf_test[1],\n",
    "            'critical_values': adf_test[4],\n",
    "            'confidence_level': 1 - threshold\n",
    "        }\n",
    "    except Exception as e:\n",
    "        print(f\"Error in stationarity test: {str(e)}\")\n",
    "        return {\n",
    "            'is_stationary': False,\n",
    "            'adf_statistic': None,\n",
    "            'p_value': None,\n",
    "            'critical_values': None,\n",
    "            'confidence_level': None\n",
    "        }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "652dc213-5faa-42c0-a163-1cd22c16ff97",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Test for Cointegration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f21da1bc-0382-4ee6-83e3-fc3bee50927f",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def test_cointegration(series1, series2, threshold=0.05, window=30):\n",
    "    \"\"\"\n",
    "    Test for cointegration with a rolling regression.\n",
    "    Uses log-transformed prices to stabilize regression.\n",
    "    Returns a single float hedge_ratio derived from median rolling parameters.\n",
    "    \"\"\"\n",
    "    # Align and drop NaN\n",
    "    series1 = series1.dropna()\n",
    "    series2 = series2.dropna()\n",
    "    common_index = series1.index.intersection(series2.index)\n",
    "    series1 = series1.loc[common_index]\n",
    "    series2 = series2.loc[common_index]\n",
    "\n",
    "    # Check for enough data\n",
    "    if len(series1) < window or len(series2) < window:\n",
    "        return None\n",
    "    \n",
    "    # Log transform prices\n",
    "    y = np.log(series1)\n",
    "    x = np.log(series2)\n",
    "    df = pd.DataFrame({'y': y, 'x': x}).dropna()\n",
    "    \n",
    "    if len(df) < window:\n",
    "        return None\n",
    "\n",
    "    try:\n",
    "        # Full sample cointegration test\n",
    "        coint_t, p_value, critical_values = coint(df['y'], df['x'])\n",
    "\n",
    "        # Add constant to x\n",
    "        X = sm.add_constant(df['x'])\n",
    "\n",
    "        if window:\n",
    "            # Rolling regression using RollingOLS\n",
    "            rols = RollingOLS(df['y'], X, window=window).fit()\n",
    "            slope = rols.params['x']\n",
    "            intercept = rols.params['const']\n",
    "            \n",
    "            # Calculate rolling R-squared\n",
    "            y_pred = slope * df['x'] + intercept\n",
    "            rolling_residuals = df['y'] - y_pred\n",
    "            rolling_tss = ((df['y'] - df['y'].rolling(window=window).mean()) ** 2).rolling(window=window).sum()\n",
    "            rolling_rss = (rolling_residuals ** 2).rolling(window=window).sum()\n",
    "            r_squared = 1 - (rolling_rss / rolling_tss)\n",
    "        else:\n",
    "            model = sm.OLS(df['y'], X).fit()\n",
    "            slope = pd.Series(model.params['x'], index=df.index)\n",
    "            intercept = pd.Series(model.params['const'], index=df.index)\n",
    "            r_squared = pd.Series(model.rsquared, index=df.index)\n",
    "\n",
    "        # Derive a single hedge ratio from median parameters\n",
    "        median_slope = slope.median()\n",
    "        median_intercept = intercept.median()\n",
    "        \n",
    "        # Median price of second series\n",
    "        median_p2 = np.median(series2)\n",
    "        if median_p2 <= 0:\n",
    "            # Avoid invalid median price\n",
    "            return None\n",
    "\n",
    "        median_ln_p2 = np.log(median_p2)\n",
    "\n",
    "        # From log model: P1 ≈ exp(intercept + slope*ln(P2))\n",
    "        # Hedge ratio ≈ (P1_median / median_p2)\n",
    "        median_hedge_ratio = np.exp(median_intercept + median_slope * median_ln_p2) / median_p2\n",
    "\n",
    "        # Test the spread based on this hedge ratio\n",
    "        spread_test = series1 - (median_hedge_ratio * series2)\n",
    "        spread_stationarity = test_stationarity(spread_test, threshold)\n",
    "\n",
    "        return {\n",
    "            'is_cointegrated': p_value < threshold and spread_stationarity['is_stationary'],\n",
    "            'coint_t_stat': coint_t,\n",
    "            'p_value': p_value,\n",
    "            'critical_values': critical_values,\n",
    "            'hedge_ratio': median_hedge_ratio,  # float hedge ratio\n",
    "            'intercept': median_intercept,       # scalar intercept if needed\n",
    "            'spread': spread_test,\n",
    "            'spread_stationarity': spread_stationarity,\n",
    "            'r_squared': r_squared\n",
    "        }\n",
    "\n",
    "    except Exception as e:\n",
    "        print(f\"Error in cointegration test: {str(e)}\")\n",
    "        return None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "191cbaa1-5b46-4546-84ec-5f7f080991b5",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Calculate Correlations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d02142a7-ab1d-450b-b209-dbb98ee1472c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def calculate_correlations(returns1, returns2, short_window=30, long_window=180):\n",
    "    \"\"\"\n",
    "    Calculate correlations using rolling windows for short and long term periods\n",
    "    \"\"\"\n",
    "    common_index = returns1.index.intersection(returns2.index)\n",
    "    returns1 = returns1[common_index]\n",
    "    returns2 = returns2[common_index]\n",
    "    \n",
    "    if len(returns1) < long_window:\n",
    "        return {\n",
    "            'short_correlation': None,\n",
    "            'long_correlation': None,\n",
    "            'correlation_divergence': None\n",
    "        }\n",
    "    \n",
    "    short_correlation = returns1.rolling(window=short_window, min_periods=short_window).corr(returns2)\n",
    "    long_correlation = returns1.rolling(window=long_window, min_periods=long_window).corr(returns2)\n",
    "    \n",
    "    correlation_divergence = long_correlation - short_correlation\n",
    "    \n",
    "    return {\n",
    "        'short_correlation': short_correlation,\n",
    "        'long_correlation': long_correlation,\n",
    "        'correlation_divergence': correlation_divergence\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "106ae0ae-4607-4211-928d-b3dae18d8362",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Mean Reversion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1b309927-ee3c-4b3e-863c-63c218b84b53",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def test_mean_reversion(series):\n",
    "    \"\"\"Test for mean reversion characteristics\"\"\"\n",
    "    if not isinstance(series, pd.Series):\n",
    "        series = pd.Series(series)\n",
    "    series = series.dropna()\n",
    "    \n",
    "    try:\n",
    "        lags = range(2, min(20, len(series)//4))\n",
    "        tau = []\n",
    "        for lag in lags:\n",
    "            lag_series = np.subtract(series[lag:].values, series[:-lag].values)\n",
    "            if len(lag_series) > 0:\n",
    "                lag_std = np.std(lag_series)\n",
    "                if not np.isnan(lag_std) and lag_std != 0:\n",
    "                    tau.append(lag_std)\n",
    "\n",
    "        if len(tau) > 1 and all(t > 0 for t in tau):\n",
    "            log_lags = np.log(list(lags)[:len(tau)])\n",
    "            log_tau = np.log(tau)\n",
    "            reg = np.polyfit(log_lags, log_tau, 1)\n",
    "            hurst = reg[0] / 2.0\n",
    "        else:\n",
    "            hurst = 0.5\n",
    "\n",
    "        df = pd.DataFrame({'y': series, 'y_lag': series.shift(1)}).dropna()\n",
    "        X = sm.add_constant(df['y_lag'])\n",
    "        model = sm.OLS(df['y'], X).fit()\n",
    "        \n",
    "        param = model.params['y_lag']\n",
    "        if 0 < param < 1:\n",
    "            half_life = -np.log(2) / np.log(param)\n",
    "            mean_reversion_speed = 1 - param\n",
    "        else:\n",
    "            half_life = np.inf\n",
    "            mean_reversion_speed = 0\n",
    "        \n",
    "        return {\n",
    "            'is_mean_reverting': bool(hurst < 0.5),\n",
    "            'hurst_exponent': float(hurst),\n",
    "            'half_life': float(half_life),\n",
    "            'mean_reversion_speed': float(mean_reversion_speed),\n",
    "            'confidence': float(model.rsquared)\n",
    "        }\n",
    "    except Exception as e:\n",
    "        print(f\"Error in mean reversion test: {str(e)}\")\n",
    "        return {\n",
    "            'is_mean_reverting': False,\n",
    "            'hurst_exponent': 0.5,\n",
    "            'half_life': np.inf,\n",
    "            'mean_reversion_speed': 0,\n",
    "            'confidence': 0\n",
    "        }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5d718a1d-707d-4e02-8083-d4b5e2be26af",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Calculate Z Score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "386f459c-6612-4d47-b95e-aa99dd0520ee",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def calculate_zscore(series, window=30):\n",
    "    \"\"\"\n",
    "    Calculate rolling z-score for a series using specified window\n",
    "    \"\"\"\n",
    "    if not isinstance(series, pd.Series):\n",
    "        series = pd.Series(series)\n",
    "    \n",
    "    rolling_mean = series.rolling(window=window).mean()\n",
    "    rolling_std = series.rolling(window=window).std()\n",
    "    \n",
    "    zscore = (series - rolling_mean) / rolling_std\n",
    "    zscore = zscore.replace([np.inf, -np.inf], np.nan)\n",
    "    \n",
    "    return zscore"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "8b640217-87bc-4554-be77-c4cedc7d734e",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Validate Pair Condition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3581297a-933f-42f8-95b8-c136fad272f3",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def validate_pair_conditions(df1, metrics, min_correlation=0.4, min_cointegration_conf=0.8):\n",
    "    \"\"\"\n",
    "    Validate if a pair meets trading conditions\n",
    "    \"\"\"\n",
    "    print(\"\\nValidation Analysis:\")\n",
    "    \n",
    "    has_enough_data = len(df1) >= 60\n",
    "    if not has_enough_data:\n",
    "        print(\"❌ Failed: Insufficient data points\")\n",
    "        return False, {'enough_data': False}\n",
    "\n",
    "    correlation = metrics['correlation']['long_correlation'].iloc[-1]\n",
    "    r_squared = metrics['cointegration']['r_squared'].iloc[-1]\n",
    "    correlation_path_valid = (correlation is not None and correlation >= min_correlation and \n",
    "                              r_squared is not None and r_squared >= 0.3)\n",
    "    \n",
    "    print(f\"Path 1 - Statistical Relationship:\")\n",
    "    print(f\"Correlation: {correlation:.4f} ({'✅' if correlation_path_valid else '❌'})\")\n",
    "    print(f\"R-squared: {r_squared:.4f} ({'✅' if r_squared >= 0.3 else '❌'})\")\n",
    "    print(f\"Path 1 Valid: {'✅' if correlation_path_valid else '❌'}\")\n",
    "\n",
    "    is_cointegrated = metrics['cointegration']['p_value'] < (1 - min_cointegration_conf)\n",
    "    is_stationary = metrics['cointegration']['spread_stationarity']['is_stationary']\n",
    "    cointegration_path_valid = is_cointegrated or is_stationary\n",
    "    \n",
    "    print(f\"\\nPath 2 - Mean Reversion Properties:\")\n",
    "    print(f\"Cointegrated: {'✅' if is_cointegrated else '❌'} (p={metrics['cointegration']['p_value']:.4f})\")\n",
    "    print(f\"Spread Stationary: {'✅' if is_stationary else '❌'} (p={metrics['cointegration']['spread_stationarity']['p_value']:.4f})\")\n",
    "    print(f\"Path 2 Valid: {'✅' if cointegration_path_valid else '❌'}\")\n",
    "\n",
    "    is_valid = has_enough_data and (correlation_path_valid or cointegration_path_valid)\n",
    "    \n",
    "    print(f\"\\nFinal Result: {'✅ PASS' if is_valid else '❌ FAIL'}\")\n",
    "    if is_valid:\n",
    "        print(\"Passed via: \" + \n",
    "              (\"Correlation/R-squared\" if correlation_path_valid else \"\") +\n",
    "              (\" and \" if correlation_path_valid and cointegration_path_valid else \"\") +\n",
    "              (\"Cointegration/Stationarity\" if cointegration_path_valid else \"\"))\n",
    "\n",
    "    conditions = {\n",
    "        'enough_data': has_enough_data,\n",
    "        'correlation_path': correlation_path_valid,\n",
    "        'cointegration_path': cointegration_path_valid\n",
    "    }\n",
    "\n",
    "    return is_valid, conditions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "31af1515-286b-4261-9a92-b07d9acdb22f",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Analyze Pair"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ea80c6b2-9053-4362-a164-8a303d1f4631",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def analyze_pair(df1, df2, config):\n",
    "    \"\"\"Analyze a pair of assets with validation\"\"\"\n",
    "    \n",
    "    if len(df1) < 60 or len(df2) < 60:\n",
    "        return None, None\n",
    "    \n",
    "    returns1 = calculate_returns(df1)\n",
    "    returns2 = calculate_returns(df2)\n",
    "    \n",
    "    correlation_metrics = calculate_correlations(returns1, returns2)\n",
    "    cointegration_metrics = test_cointegration(df1['close'], df2['close'])\n",
    "    \n",
    "    if cointegration_metrics is None or not cointegration_metrics['is_cointegrated']:\n",
    "        return None, None\n",
    "    \n",
    "    # Check hedge ratio reasonableness\n",
    "    hedge_ratio = cointegration_metrics['hedge_ratio']\n",
    "    if hedge_ratio is None or hedge_ratio <= 0:\n",
    "        return None, None\n",
    "    \n",
    "    # If hedge ratio is extremely large or small, skip\n",
    "    if hedge_ratio > 10 or hedge_ratio < 0.1:\n",
    "        print(f\"Hedge ratio out of reasonable bounds: {hedge_ratio:.4f}, skipping this pair.\")\n",
    "        return None, None\n",
    "    \n",
    "    spread = cointegration_metrics['spread']\n",
    "    mr_metrics = test_mean_reversion(spread)\n",
    "    \n",
    "    metrics = {\n",
    "        'correlation': correlation_metrics,\n",
    "        'cointegration': cointegration_metrics,\n",
    "        'mean_reversion': mr_metrics\n",
    "    }\n",
    "    \n",
    "    is_valid, conditions = validate_pair_conditions(df1, metrics)\n",
    "    if not is_valid:\n",
    "        return None, None\n",
    "    \n",
    "    series_data = pd.DataFrame({\n",
    "        'pair': f\"{df1['instrument'].iloc[0]}/{df2['instrument'].iloc[0]}\",\n",
    "        'exchange': df1['exchange'].iloc[0],\n",
    "        'price1': df1['close'],\n",
    "        'price2': df2['close'],\n",
    "        'returns1': returns1,\n",
    "        'returns2': returns2,\n",
    "        'hedge_ratio': hedge_ratio,\n",
    "        'spread': spread,\n",
    "        'spread_zscore': calculate_zscore(spread),\n",
    "        'correlation': correlation_metrics['long_correlation']\n",
    "    })\n",
    "    \n",
    "    return series_data, metrics"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0f618712-e2b7-4143-b921-76c16d29e02b",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Generate Signals"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "98c980d3-4800-4f1f-9ea4-c5a4ceb73b44",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def generate_signals(series_data, metrics, config):\n",
    "    \"\"\"\n",
    "    Generate trading signals with proper type handling and position-based exits\n",
    "    \n",
    "    Parameters:\n",
    "    series_data (pd.DataFrame): Input data containing spread_zscore\n",
    "    metrics: Additional metrics (unused in current version)\n",
    "    config (dict): Configuration parameters including zscore_threshold\n",
    "    \n",
    "    1 = buy/long entry\n",
    "    -1 = sell/short entry\n",
    "    0 = no trade\n",
    "    -2 = exit long position (sell to close)\n",
    "    2 = exit short position (buy to cover)\n",
    "    \"\"\"\n",
    "    signals = pd.DataFrame(index=series_data.index)\n",
    "    zscore = series_data['spread_zscore']\n",
    "    zscore_threshold = config['zscore_threshold']\n",
    "    \n",
    "    signals['signal'] = 0\n",
    "    signals['zscore'] = zscore\n",
    "    current_position = 0\n",
    "    \n",
    "    for i in range(1, len(signals)):\n",
    "        curr_zscore = zscore.iloc[i]\n",
    "        prev_zscore = zscore.iloc[i-1]\n",
    "        \n",
    "        if pd.isna(curr_zscore) or pd.isna(prev_zscore):\n",
    "            continue\n",
    "            \n",
    "        if current_position == 0:\n",
    "            if curr_zscore > zscore_threshold and prev_zscore <= zscore_threshold:\n",
    "                signals.iloc[i, signals.columns.get_loc('signal')] = -1\n",
    "                current_position = -1\n",
    "            elif curr_zscore < -zscore_threshold and prev_zscore >= -zscore_threshold:\n",
    "                signals.iloc[i, signals.columns.get_loc('signal')] = 1\n",
    "                current_position = 1\n",
    "        elif (curr_zscore < 0 and prev_zscore >= 0) or (curr_zscore > 0 and prev_zscore <= 0):\n",
    "            if current_position == 1:\n",
    "                signals.iloc[i, signals.columns.get_loc('signal')] = -2\n",
    "                current_position = 0\n",
    "            elif current_position == -1:\n",
    "                signals.iloc[i, signals.columns.get_loc('signal')] = 2\n",
    "                current_position = 0\n",
    "    \n",
    "    signals['signal'] = signals['signal'].fillna(0)\n",
    "    return signals"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a3f5b7d1-68d7-449e-bf3c-36c2211eb67b",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Calculate Performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dc17942a-cc90-4b90-bfed-2bb4ee2c57c7",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def calculate_performance_metrics(returns, trades_df, cumulative_returns):\n",
    "    \"\"\"Calculate performance metrics\"\"\"\n",
    "    total_return = cumulative_returns.iloc[-1] - 1\n",
    "    n_years = len(returns) / 365\n",
    "    if total_return > -1:\n",
    "        annual_return = (1 + total_return) ** (1/n_years) - 1\n",
    "    else:\n",
    "        annual_return = total_return / n_years\n",
    "        \n",
    "    daily_volatility = returns.std()\n",
    "    annual_volatility = daily_volatility * np.sqrt(365)\n",
    "    sharpe_ratio = annual_return / annual_volatility if annual_volatility != 0 else 0\n",
    "    \n",
    "    rolling_max = cumulative_returns.expanding().max()\n",
    "    drawdowns = (cumulative_returns - rolling_max) / rolling_max\n",
    "    max_drawdown = drawdowns.min()\n",
    "    \n",
    "    if len(trades_df) > 0:\n",
    "        trade_returns = trades_df['return']\n",
    "        winning_trades = trades_df[trade_returns > 0]\n",
    "        losing_trades = trades_df[trade_returns <= 0]\n",
    "        \n",
    "        win_rate = len(winning_trades) / len(trades_df) if len(trades_df) > 0 else 0\n",
    "        avg_win = winning_trades['return'].mean() if len(winning_trades) > 0 else 0\n",
    "        avg_loss = losing_trades['return'].mean() if len(losing_trades) > 0 else 0\n",
    "        profit_factor = abs(winning_trades['return'].sum() / losing_trades['return'].sum()) if (len(losing_trades) > 0 and losing_trades['return'].sum() != 0) else 0\n",
    "        avg_return_per_trade = trade_returns.mean() if len(trade_returns) > 0 else 0\n",
    "    else:\n",
    "        win_rate = 0\n",
    "        avg_win = 0\n",
    "        avg_loss = 0\n",
    "        profit_factor = 0\n",
    "        avg_return_per_trade = 0\n",
    "    \n",
    "    return {\n",
    "        'total_return': total_return,\n",
    "        'annual_return': annual_return,\n",
    "        'annual_volatility': annual_volatility,\n",
    "        'sharpe_ratio': sharpe_ratio,\n",
    "        'max_drawdown': max_drawdown,\n",
    "        'num_trades': len(trades_df),\n",
    "        'win_rate': win_rate,\n",
    "        'avg_win': avg_win,\n",
    "        'avg_loss': avg_loss,\n",
    "        'profit_factor': profit_factor,\n",
    "        'avg_return_per_trade': avg_return_per_trade\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ff6f7df1-7867-4ffe-a69e-a03e6b8c4be0",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Backtest Pair Strategy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2c007f45-871f-48f2-9740-3832e8c1f082",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def backtest_pair_strategy(series_data, signals, config):\n",
    "    \"\"\"\n",
    "    Backtest pairs trading strategy with proper position tracking, stop loss, and profit taking\n",
    "    \"\"\"\n",
    "    dtypes = {\n",
    "        'position': 'int32',\n",
    "        'equity': 'float64',\n",
    "        'returns': 'float64',\n",
    "        'unrealized_pnl': 'float64',\n",
    "        'active_qty1': 'float64',\n",
    "        'active_qty2': 'float64'\n",
    "    }\n",
    "    \n",
    "    result_df = pd.DataFrame(\n",
    "        {\n",
    "            'position': [0]*len(signals),\n",
    "            'equity': [0.0]*len(signals),\n",
    "            'returns': [0.0]*len(signals),\n",
    "            'unrealized_pnl': [0.0]*len(signals),\n",
    "            'active_qty1': [0.0]*len(signals),\n",
    "            'active_qty2': [0.0]*len(signals)\n",
    "        }, index=signals.index\n",
    "    ).astype(dtypes)\n",
    "    \n",
    "    position = 0\n",
    "    running_equity = config['initial_capital']  # Realized equity\n",
    "    returns = []\n",
    "    trades = []\n",
    "    active_qty1 = 0.0\n",
    "    active_qty2 = 0.0\n",
    "    entry_price1 = 0.0\n",
    "    entry_price2 = 0.0\n",
    "    entry_date = None\n",
    "    equity_curve = []\n",
    "\n",
    "    prev_daily_equity = running_equity  # Used to calculate daily returns\n",
    "\n",
    "    for i in range(len(signals)):\n",
    "        current_signal = signals['signal'].iloc[i]\n",
    "        price1 = series_data['price1'].iloc[i]\n",
    "        price2 = series_data['price2'].iloc[i]\n",
    "        hedge_ratio = series_data['hedge_ratio'].iloc[i]\n",
    "        current_date = signals.index[i]\n",
    "        \n",
    "        # Calculate current PnL (unrealized)\n",
    "        if position == 1:\n",
    "            current_pnl = (active_qty1 * (price1 - entry_price1)) - (active_qty2 * (price2 - entry_price2))\n",
    "        elif position == -1:\n",
    "            current_pnl = (-active_qty1 * (price1 - entry_price1)) + (active_qty2 * (price2 - entry_price2))\n",
    "        else:\n",
    "            current_pnl = 0.0\n",
    "        \n",
    "        period_return = 0.0\n",
    "        \n",
    "        # Check stop loss / take profit if in a position\n",
    "        if position != 0:\n",
    "            pnl_pct = current_pnl / running_equity if running_equity != 0 else 0\n",
    "            if pnl_pct <= -config['stop_loss_pct']:\n",
    "                exit_type = 'stop_loss'\n",
    "            elif pnl_pct >= config['take_profit_pct']:\n",
    "                exit_type = 'take_profit'\n",
    "            else:\n",
    "                exit_type = None\n",
    "            \n",
    "            if exit_type:\n",
    "                total_cost = (abs(active_qty1*price1) + abs(active_qty2*price2)) * config['transaction_cost']\n",
    "                trade_return = current_pnl - total_cost\n",
    "                running_equity += trade_return  # Realize the trade return\n",
    "                # After realizing trade, current_pnl = 0 since position closes\n",
    "                current_pnl = 0.0\n",
    "\n",
    "                trades.append({\n",
    "                    'date': current_date,\n",
    "                    'entry_date': entry_date,\n",
    "                    'signal': exit_type,\n",
    "                    'entry_price1': entry_price1,\n",
    "                    'entry_price2': entry_price2,\n",
    "                    'exit_price1': price1,\n",
    "                    'exit_price2': price2,\n",
    "                    'qty1': active_qty1,\n",
    "                    'qty2': active_qty2,\n",
    "                    'return': trade_return,\n",
    "                    'costs': total_cost,\n",
    "                    'position': position,\n",
    "                    'pnl_pct': pnl_pct * 100\n",
    "                })\n",
    "\n",
    "                position = 0\n",
    "                active_qty1 = 0.0\n",
    "                active_qty2 = 0.0\n",
    "                entry_date = None\n",
    "        \n",
    "        # Handle new entries\n",
    "        elif current_signal in [1, -1] and position == 0:\n",
    "            position_size = running_equity * config['position_size_pct']\n",
    "            \n",
    "            if price1 <= 0 or price2 <= 0 or hedge_ratio is None or hedge_ratio <= 0:\n",
    "                # Invalid conditions for a trade\n",
    "                pass\n",
    "            else:\n",
    "                active_qty1 = position_size / price1\n",
    "                attempted_qty2 = (position_size * hedge_ratio) / price2\n",
    "                \n",
    "                if attempted_qty2 > 1e6:\n",
    "                    print(\"Attempted qty2 too large, skipping trade.\")\n",
    "                else:\n",
    "                    active_qty2 = attempted_qty2\n",
    "                    entry_price1 = price1\n",
    "                    entry_price2 = price2\n",
    "                    entry_date = current_date\n",
    "                    total_cost = (abs(active_qty1*price1) + abs(active_qty2*price2)) * config['transaction_cost']\n",
    "                    running_equity -= total_cost\n",
    "                    position = 1 if current_signal == 1 else -1\n",
    "\n",
    "        # Handle normal exits\n",
    "        elif ((current_signal == -2 and position == 1) or \n",
    "              (current_signal == 2 and position == -1)):\n",
    "            \n",
    "            total_cost = (abs(active_qty1*price1) + abs(active_qty2*price2)) * config['transaction_cost']\n",
    "            trade_return = current_pnl - total_cost\n",
    "            running_equity += trade_return\n",
    "            current_pnl = 0.0  # Position closed, unrealized pnl realized\n",
    "            trades.append({\n",
    "                'date': current_date,\n",
    "                'entry_date': entry_date,\n",
    "                'signal': current_signal,\n",
    "                'entry_price1': entry_price1,\n",
    "                'entry_price2': entry_price2,\n",
    "                'exit_price1': price1,\n",
    "                'exit_price2': price2,\n",
    "                'qty1': active_qty1,\n",
    "                'qty2': active_qty2,\n",
    "                'return': trade_return,\n",
    "                'costs': total_cost,\n",
    "                'position': position,\n",
    "                'pnl_pct': (trade_return / running_equity)*100 if running_equity!=0 else 0\n",
    "            })\n",
    "\n",
    "            position = 0\n",
    "            active_qty1 = 0.0\n",
    "            active_qty2 = 0.0\n",
    "            entry_date = None\n",
    "        \n",
    "        # Daily mark-to-market equity calculation\n",
    "        daily_equity = running_equity + current_pnl\n",
    "        \n",
    "        # Calculate daily returns based on daily_equity changes\n",
    "        if prev_daily_equity != 0:\n",
    "            period_return = (daily_equity / prev_daily_equity) - 1\n",
    "        else:\n",
    "            period_return = 0.0\n",
    "        \n",
    "        prev_daily_equity = daily_equity\n",
    "        \n",
    "        result_df.loc[signals.index[i]] = {\n",
    "            'position': position,\n",
    "            'equity': daily_equity,\n",
    "            'returns': period_return,\n",
    "            'unrealized_pnl': current_pnl,\n",
    "            'active_qty1': active_qty1,\n",
    "            'active_qty2': active_qty2\n",
    "        }\n",
    "        \n",
    "        returns.append(period_return)\n",
    "        equity_curve.append(daily_equity)\n",
    "\n",
    "    returns = pd.Series(returns, index=signals.index)\n",
    "    trades_df = pd.DataFrame(trades)\n",
    "    cumulative_returns = pd.Series(equity_curve, index=signals.index) / config['initial_capital']\n",
    "    \n",
    "    performance = calculate_performance_metrics(returns, trades_df, cumulative_returns)\n",
    "    \n",
    "    return result_df, returns, performance, trades_df, cumulative_returns\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b8cc41f8-a600-4214-882c-6bf3a6889212",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Run Pair Trading System"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "44fb9479-7728-40a9-918d-bfe5784404a2",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def run_pair_trading_system(prices_df, pairs, exchanges, config):\n",
    "    \"\"\"Run complete pair trading analysis system\"\"\"\n",
    "    results = []\n",
    "    \n",
    "    if 'exchangeTimestamp' in prices_df.columns and not isinstance(prices_df.index, pd.DatetimeIndex):\n",
    "        prices_df = prices_df.set_index('exchangeTimestamp')\n",
    "    \n",
    "    for exchange in exchanges:\n",
    "        for i, pair1 in enumerate(pairs):\n",
    "            for pair2 in pairs[i+1:]:\n",
    "                print(f\"\\nAnalyzing {pair1}/{pair2} on {exchange}\")\n",
    "                \n",
    "                df1 = prices_df[(prices_df['instrument'] == pair1) & (prices_df['exchange'] == exchange)].copy()\n",
    "                df2 = prices_df[(prices_df['instrument'] == pair2) & (prices_df['exchange'] == exchange)].copy()\n",
    "                \n",
    "                df1 = df1.sort_index()\n",
    "                df2 = df2.sort_index()\n",
    "\n",
    "                common_index = df1.index.intersection(df2.index)\n",
    "                df1 = df1.loc[common_index]\n",
    "                df2 = df2.loc[common_index]\n",
    "                \n",
    "                df1 = df1.dropna(subset=['close'])\n",
    "                df2 = df2.dropna(subset=['close'])\n",
    "                \n",
    "                if len(df1) < config['min_data_points'] or len(df2) < config['min_data_points']:\n",
    "                    print(f\"Insufficient data: {pair1}: {len(df1)} days, {pair2}: {len(df2)} days\")\n",
    "                    continue\n",
    "                \n",
    "                if df1['close'].isnull().any() or df2['close'].isnull().any():\n",
    "                    print(f\"Found missing prices in {pair1} or {pair2}\")\n",
    "                    continue\n",
    "                \n",
    "                try:\n",
    "                    print(f\"Data ranges:\")\n",
    "                    print(f\"{pair1}: {df1.index.min()} to {df1.index.max()}\")\n",
    "                    print(f\"{pair2}: {df2.index.min()} to {df2.index.max()}\")\n",
    "                    print(f\"Number of records: {pair1}: {len(df1)}, {pair2}: {len(df2)}\")\n",
    "                    \n",
    "                    series_data, metrics = analyze_pair(df1, df2, config)\n",
    "                    if series_data is None or metrics is None:\n",
    "                        print(\"Pair failed statistical validation\")\n",
    "                        continue\n",
    "                    \n",
    "                    print(f\"Cointegration p-value: {metrics['cointegration']['p_value']:.4f}\")\n",
    "                    print(f\"Hedge ratio: {metrics['cointegration']['hedge_ratio']:.4f}\")\n",
    "                    print(f\"Mean reversion - Hurst: {metrics['mean_reversion']['hurst_exponent']:.4f}\")\n",
    "                    \n",
    "                    signals = generate_signals(series_data, metrics, config)\n",
    "                    \n",
    "                    result_df, ret_series, performance, trades, equity_curve = backtest_pair_strategy(\n",
    "                        series_data, signals, config\n",
    "                    )\n",
    "                    \n",
    "                    if performance['num_trades'] > 0:\n",
    "                        print(f\"Found {performance['num_trades']} trades\")\n",
    "                        print(f\"Total return: {performance['total_return']:.2%}\")\n",
    "                        print(f\"Sharpe ratio: {performance['sharpe_ratio']:.2f}\")\n",
    "                        print(f\"Win rate: {performance['win_rate']:.2%}\")\n",
    "                        \n",
    "                        result = {\n",
    "                            'pair': f\"{pair1}/{pair2}\",\n",
    "                            'exchange': exchange,\n",
    "                            'series_data': series_data,\n",
    "                            'metrics': metrics,\n",
    "                            'signals': signals,\n",
    "                            'returns': ret_series,\n",
    "                            'performance': performance,\n",
    "                            'trades': trades,\n",
    "                            'equity_curve': equity_curve,\n",
    "                            'result_df':result_df\n",
    "                        }\n",
    "                        results.append(result)\n",
    "                    else:\n",
    "                        print(\"No trades executed\")\n",
    "                        \n",
    "                except Exception as e:\n",
    "                    print(f\"Error analyzing pair {pair1}/{pair2}: {str(e)}\")\n",
    "                    import traceback\n",
    "                    print(traceback.format_exc())\n",
    "                    continue\n",
    "    \n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "cf9e92c7-55d2-4c71-9b16-0832c352ca55",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "##Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "625834c8-1058-496f-bbdd-e1a032f4d77c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "pairs = [\n",
    "    \"btc_usdt\",\n",
    "    \"eth_usdt\",\n",
    "    \"sol_usdt\",\n",
    "    \"xrp_usdt\",\n",
    "    \"ada_usdt\",\n",
    "    \"avax_usdt\",\n",
    "    \"dot_usdt\",\n",
    "    \"doge_usdt\",\n",
    "    \"link_usdt\",\n",
    "    \"matic_usdt\",\n",
    "    \"shib_usdt\",\n",
    "    \"ltc_usdt\",\n",
    "    \"uni_usdt\",\n",
    "    \"atom_usdt\",\n",
    "    \"etc_usdt\",\n",
    "    \"fil_usdt\",\n",
    "    \"near_usdt\",\n",
    "    \"algo_usdt\",\n",
    "    \"vet_usdt\",\n",
    "    \"apt_usdt\",\n",
    "    \"sand_usdt\",\n",
    "    \"mana_usdt\",\n",
    "    \"gala_usdt\",\n",
    "    \"aave_usdt\"\n",
    "]\n",
    "\n",
    "exchanges = ['bitfinex']\n",
    "\n",
    "today = datetime.today()\n",
    "start_date = '01-01-2020'\n",
    "end_date = today.strftime('%d-%m-%Y')\n",
    "\n",
    "print(\"Fetching price data...\")\n",
    "prices_df = fetch_crypto_prices(pairs, exchanges, start_date, end_date)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "28e4f278-8e15-460e-a060-125594163727",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Configuration and Execute Search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "37c98a52-60b8-4319-94f0-e69c20ef5585",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "config = {\n",
    "    'initial_capital': 1000000,\n",
    "    'position_size_pct': 0.05,\n",
    "    'transaction_cost': 0.001, \n",
    "    'stop_loss_pct': 0.05,\n",
    "    'take_profit_pct': 0.30,\n",
    "    'min_correlation': 0.3,\n",
    "    'min_cointegration_conf': 0.90,\n",
    "    'zscore_threshold': 3.0,\n",
    "    'min_data_points': 60\n",
    "}\n",
    "\n",
    "results = run_pair_trading_system(\n",
    "    prices_df=prices_df,\n",
    "    pairs=pairs,\n",
    "    exchanges=exchanges,\n",
    "    config=config\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "396b278e-52cc-415c-827f-e0aad6a4bbf5",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Print Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "fc226213-4933-4f3b-8102-caa75876b384",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def print_trading_summary(results):\n",
    "    print(\"\\n=== Pair Trading Analysis Summary ===\")\n",
    "    for result in results:\n",
    "        print(f\"\\nPair: {result['pair']} on {result['exchange']}\")\n",
    "        metrics = result['metrics']\n",
    "        perf = result['performance']\n",
    "        \n",
    "        print(\"\\nStatistical Properties:\")\n",
    "        print(f\"Cointegration p-value: {metrics['cointegration']['p_value']:.4f}\")\n",
    "        print(f\"Hedge ratio: {metrics['cointegration']['hedge_ratio']:.4f}\")\n",
    "        print(f\"Long-term correlation: {metrics['correlation']['long_correlation'].iloc[-1]:.4f}\")\n",
    "        print(f\"Hurst exponent: {metrics['mean_reversion']['hurst_exponent']:.4f}\")\n",
    "        \n",
    "        print(\"\\nTrading Performance:\")\n",
    "        print(f\"Total Return: {perf['total_return']*100:.2f}%\")\n",
    "        print(f\"Annual Return: {perf['annual_return']*100:.2f}%\")\n",
    "        print(f\"Sharpe Ratio: {perf['sharpe_ratio']:.2f}\")\n",
    "        print(f\"Max Drawdown: {perf['max_drawdown']*100:.2f}%\")\n",
    "        print(f\"Number of Trades: {perf['num_trades']}\")\n",
    "        print(f\"Win Rate: {perf['win_rate']*100:.2f}%\")\n",
    "        print(f\"Profit Factor: {perf['profit_factor']:.2f}\")\n",
    "        print(\"\\n\" + \"=\"*50)\n",
    "\n",
    "def print_winning_trading_summary(results):\n",
    "    winners_found = False\n",
    "    print(\"\\n=== Winning Pairs Trading Analysis Summary ===\")\n",
    "    \n",
    "    for result in results:\n",
    "        perf = result['performance']\n",
    "        if (perf['total_return'] > 0 and \n",
    "            perf['sharpe_ratio'] > 0 and \n",
    "            perf['win_rate'] > 0.2):\n",
    "            winners_found = True\n",
    "            print(f\"\\nPair: {result['pair']} on {result['exchange']}\")\n",
    "            metrics = result['metrics']\n",
    "            \n",
    "            print(\"\\nStatistical Properties:\")\n",
    "            print(f\"Cointegration p-value: {metrics['cointegration']['p_value']:.4f}\")\n",
    "            print(f\"Hedge ratio: {metrics['cointegration']['hedge_ratio']:.4f}\")\n",
    "            print(f\"Long-term correlation: {metrics['correlation']['long_correlation'].iloc[-1]:.4f}\")\n",
    "            print(f\"Hurst exponent: {metrics['mean_reversion']['hurst_exponent']:.4f}\")\n",
    "            \n",
    "            print(\"\\nTrading Performance:\")\n",
    "            print(f\"Total Return: {perf['total_return']*100:.2f}%\")\n",
    "            print(f\"Annual Return: {perf['annual_return']*100:.2f}%\")\n",
    "            print(f\"Sharpe Ratio: {perf['sharpe_ratio']:.2f}\")\n",
    "            print(f\"Max Drawdown: {perf['max_drawdown']*100:.2f}%\")\n",
    "            print(f\"Number of Trades: {perf['num_trades']}\")\n",
    "            print(f\"Win Rate: {perf['win_rate']*100:.2f}%\")\n",
    "            print(f\"Profit Factor: {perf['profit_factor']:.2f}\")\n",
    "            print(\"\\n\" + \"=\"*50)\n",
    "    \n",
    "    if not winners_found:\n",
    "        print(\"\\nNo winning pairs found matching the criteria.\")\n",
    "        print(\"Criteria for winners:\")\n",
    "        print(\"- Positive total return\")\n",
    "        print(\"- Positive Sharpe ratio\")\n",
    "        print(\"- Win rate above 20%\")\n",
    "        print(\"\\n\" + \"=\"*50)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "df08c5c7-d61c-48ce-ba00-bdb23bc9f33a",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Print results\n",
    "print_trading_summary(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a935ed22-ae7f-46a6-b364-3ab30700e656",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Print winning trade summary\n",
    "print_winning_trading_summary(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "6bb3fca6-ceb8-42dd-98e9-5fe20de3937c",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "## Plot Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "f68e959d-b173-4a1b-8eb6-d9e18898aae3",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def plot_enhanced_result(result, config):\n",
    "    \"\"\"\n",
    "    Plot a comprehensive set of charts for a single result record:\n",
    "    - Spread and trades\n",
    "    - Z-score of spread and trades\n",
    "    - Equity curve\n",
    "    - Rolling correlations (short and long term)\n",
    "    - Rolling R-squared from cointegration\n",
    "    - Mark stop loss exits with a black 'X' and take profit exits with a black '+'\n",
    "    \"\"\"\n",
    "\n",
    "    series_data = result['series_data']\n",
    "    signals = result['signals']\n",
    "    trades = result['trades']\n",
    "    result_df = result['result_df']\n",
    "    metrics = result['metrics']\n",
    "\n",
    "    spread = series_data['spread']\n",
    "    zscore = series_data['spread_zscore']\n",
    "    equity = result_df['equity'] if 'equity' in result_df.columns else None\n",
    "\n",
    "    # Identify entry and exit signals\n",
    "    long_entries = signals[signals['signal'] == 1].index\n",
    "    short_entries = signals[signals['signal'] == -1].index\n",
    "    long_exits = signals[signals['signal'] == -2].index\n",
    "    short_exits = signals[signals['signal'] == 2].index\n",
    "\n",
    "    # Extract correlation data if available\n",
    "    short_corr = metrics['correlation'].get('short_correlation')\n",
    "    long_corr = metrics['correlation'].get('long_correlation')\n",
    "\n",
    "    # Extract rolling R-squared if available\n",
    "    r_squared = metrics['cointegration'].get('r_squared')\n",
    "\n",
    "    # Mean reversion metrics\n",
    "    mr_metrics = metrics['mean_reversion']\n",
    "    hurst = mr_metrics.get('hurst_exponent', None)\n",
    "    half_life = mr_metrics.get('half_life', None)\n",
    "    mean_reversion_speed = mr_metrics.get('mean_reversion_speed', None)\n",
    "    confidence = mr_metrics.get('confidence', None)\n",
    "\n",
    "    # We have 5 subplots: Spread, Z-score, Equity, Correlations, R-squared\n",
    "    fig, axes = plt.subplots(5, 1, figsize=(14, 16), sharex=True)\n",
    "\n",
    "    fig.suptitle(f\"{result['pair']} on {result['exchange']} - Comprehensive Analysis\",\n",
    "                 fontsize=14, y=0.98)\n",
    "\n",
    "    # --- Subplot 1: Spread with trades ---\n",
    "    ax1 = axes[0]\n",
    "    ax1.plot(spread.index, spread, label='Spread', color='blue')\n",
    "    ax1.scatter(long_entries, spread.loc[long_entries], marker='^', color='green', s=100, label='Long Entry')\n",
    "    ax1.scatter(short_entries, spread.loc[short_entries], marker='v', color='red', s=100, label='Short Entry')\n",
    "    ax1.scatter(long_exits, spread.loc[long_exits], marker='x', color='green', s=100, label='Long Exit')\n",
    "    ax1.scatter(short_exits, spread.loc[short_exits], marker='x', color='red', s=100, label='Short Exit')\n",
    "\n",
    "    # Mark stop_loss and take_profit exits:\n",
    "    # We'll loop through the trades DataFrame\n",
    "    for _, trade in trades.iterrows():\n",
    "        exit_signal = trade['signal']  # could be 'stop_loss', 'take_profit', or numeric\n",
    "        exit_date = trade['date']\n",
    "        if exit_date in spread.index:  # Ensure we have data for that date\n",
    "            if exit_signal == 'stop_loss':\n",
    "                ax1.scatter(exit_date, spread.loc[exit_date], marker='x', color='black', s=100, label='Stop Loss Exit')\n",
    "            elif exit_signal == 'take_profit':\n",
    "                ax1.scatter(exit_date, spread.loc[exit_date], marker='+', color='black', s=100, label='Take Profit Exit')\n",
    "\n",
    "    ax1.set_ylabel('Spread')\n",
    "    ax1.set_title('Spread with Trade Signals', fontsize=12)\n",
    "    # We may have multiple legends due to repeated labels. Let's handle that:\n",
    "    handles, labels = ax1.get_legend_handles_labels()\n",
    "    # Remove duplicates while preserving order\n",
    "    unique = list(dict(zip(labels, handles)).items())\n",
    "    ax1.legend([u[1] for u in unique],[u[0] for u in unique], fontsize=9, loc='best')\n",
    "    ax1.grid(True)\n",
    "\n",
    "    # --- Subplot 2: Z-score with thresholds and trades ---\n",
    "    ax2 = axes[1]\n",
    "    ax2.plot(zscore.index, zscore, label='Z-score', color='purple')\n",
    "    threshold = config['zscore_threshold']\n",
    "    ax2.axhline(y=threshold, color='red', linestyle='--', linewidth=1, label=f'+{threshold} Threshold')\n",
    "    ax2.axhline(y=-threshold, color='red', linestyle='--', linewidth=1, label=f'-{threshold} Threshold')\n",
    "    ax2.axhline(y=0, color='black', linestyle='-', linewidth=1, label='Zero Line')\n",
    "    ax2.scatter(long_entries, zscore.loc[long_entries], marker='^', color='green', s=100)\n",
    "    ax2.scatter(short_entries, zscore.loc[short_entries], marker='v', color='red', s=100)\n",
    "    ax2.scatter(long_exits, zscore.loc[long_exits], marker='x', color='green', s=100)\n",
    "    ax2.scatter(short_exits, zscore.loc[short_exits], marker='x', color='red', s=100)\n",
    "\n",
    "    # No special markers for stop_loss or take_profit here, just on spread chart\n",
    "    ax2.set_ylabel('Z-score')\n",
    "    ax2.set_title('Z-score of Spread', fontsize=12)\n",
    "    ax2.legend(loc='best', fontsize=9)\n",
    "    ax2.grid(True)\n",
    "\n",
    "    # --- Subplot 3: Equity Curve ---\n",
    "    ax3 = axes[2]\n",
    "    if equity is not None:\n",
    "        ax3.plot(equity.index, equity, label='Equity', color='blue')\n",
    "        ax3.set_ylabel('Equity')\n",
    "        ax3.set_title('Equity Curve', fontsize=12)\n",
    "        ax3.legend(loc='best', fontsize=9)\n",
    "        ax3.grid(True)\n",
    "    else:\n",
    "        ax3.text(0.5, 0.5, 'No Equity Data Available', ha='center', va='center',\n",
    "                 transform=ax3.transAxes, fontsize=10)\n",
    "\n",
    "    # --- Subplot 4: Rolling Correlations ---\n",
    "    ax4 = axes[3]\n",
    "    if short_corr is not None and long_corr is not None:\n",
    "        ax4.plot(short_corr.index, short_corr, label='Short-term Correlation', color='orange')\n",
    "        ax4.plot(long_corr.index, long_corr, label='Long-term Correlation', color='green')\n",
    "        ax4.set_ylabel('Correlation')\n",
    "        ax4.set_title('Rolling Correlations', fontsize=12)\n",
    "        ax4.legend(loc='best', fontsize=9)\n",
    "        ax4.grid(True)\n",
    "    else:\n",
    "        ax4.text(0.5, 0.5, 'No Rolling Correlation Data Available', ha='center', va='center',\n",
    "                 transform=ax4.transAxes, fontsize=10)\n",
    "\n",
    "    # --- Subplot 5: Rolling R-squared ---\n",
    "    ax5 = axes[4]\n",
    "    if r_squared is not None:\n",
    "        ax5.plot(r_squared.index, r_squared, label='Rolling R-squared', color='magenta')\n",
    "        ax5.set_ylabel('R-squared')\n",
    "        ax5.set_title('Rolling R-squared of Cointegration Regression', fontsize=12)\n",
    "        ax5.legend(loc='best', fontsize=9)\n",
    "        ax5.grid(True)\n",
    "    else:\n",
    "        ax5.text(0.5, 0.5, 'No Rolling R-squared Data Available', ha='center', va='center',\n",
    "                 transform=ax5.transAxes, fontsize=10)\n",
    "\n",
    "    for ax in axes:\n",
    "        ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d'))\n",
    "        ax.tick_params(axis='x', rotation=45, labelsize=9)\n",
    "\n",
    "    plt.tight_layout()\n",
    "    plt.show()\n",
    "\n",
    "    \n",
    "for r in results[:3]:\n",
    "    plot_enhanced_result(r, config) "
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "environmentMetadata": {
    "base_environment": "",
    "client": "1"
   },
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "Cointegration Based Crypto Pairs Trading Strategy",
   "widgets": {}
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}