{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 📘 Fincept Notebook — Pairs Trading: Spread Z-Score\n",
        "\n",
        "**Quant · Hard · ~32 min · pandas + numpy**\n",
        "\n",
        "Pairs trading is the classic market-neutral, mean-reversion strategy. The idea: find two assets that move together (driven by a shared factor), trade the *spread* between them rather than either one outright, and bet that the spread reverts to its average. When the spread stretches too far from its mean we fade it — short the rich leg, long the cheap leg — and close when it snaps back. Because we hold offsetting long/short legs, the strategy is largely immune to the overall market direction.\n",
        "\n",
        "**What you'll learn**\n",
        "- How to generate two cointegrated-ish series from a shared stochastic trend\n",
        "- How to estimate the hedge ratio β by least squares and build a stationary spread\n",
        "- How to turn a rolling z-score into entry/exit signals and backtest the P&L (trades, hit rate, return, Sharpe)\n",
        "\n",
        "> **Requires** `pandas` and `numpy` (bundled with Fincept Terminal's Python environment).\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 1. Build two cointegrated price series\n",
        "\n",
        "We simulate a **shared stochastic trend** `F` (a random walk — the common factor, like a sector index) and give each asset its own loading on that factor plus **idiosyncratic noise**. Because both prices are driven by the same `F`, their spread is *stationary* (mean-reverting) even though each price wanders. This is the structural setup behind cointegration. All generation is deterministic given `np.random.seed(7)`, so every cell reproduces the identical series.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "fincept_title": "Generate Pair"
      },
      "outputs": [],
      "source": [
        "try:\n",
        "    import numpy as np\n",
        "    import pandas as pd\n",
        "except ImportError:\n",
        "    raise SystemExit(\"Fincept Notebook needs pandas + numpy — install with: pip install pandas numpy\")\n",
        "\n",
        "np.random.seed(7)\n",
        "\n",
        "N = 750   # ~3 years of trading days\n",
        "\n",
        "# Shared stochastic trend (common factor) = random walk\n",
        "factor = np.cumsum(np.random.normal(0.0, 1.0, N)) + 100.0\n",
        "\n",
        "# Asset A and B load on the factor with different betas + idiosyncratic noise.\n",
        "# A mean-reverting spread component is injected so the pair is genuinely cointegrated.\n",
        "noise_a = np.random.normal(0.0, 0.6, N)\n",
        "noise_b = np.random.normal(0.0, 0.6, N)\n",
        "\n",
        "A = 1.0 * factor + 20.0 + noise_a\n",
        "B = 0.6 * factor + 50.0 + noise_b\n",
        "\n",
        "prices = pd.DataFrame({\"A\": A, \"B\": B})\n",
        "\n",
        "print(f\"Generated {N} days of two cointegrated-ish prices (seed=7)\")\n",
        "print()\n",
        "print(\"Head:\")\n",
        "print(prices.head().round(4).to_string())\n",
        "print()\n",
        "print(\"Describe:\")\n",
        "print(prices.describe().round(4).to_string())\n",
        "print()\n",
        "print(f\"Correlation of A and B levels: {prices['A'].corr(prices['B']):.4f}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 2. Estimate the hedge ratio β\n",
        "\n",
        "We can't just trade `A − B`; the legs must be *dollar-balanced* in their factor exposure. We regress `A` on `B` by ordinary least squares to get the **hedge ratio β** (and an intercept α):\n",
        "\n",
        "$$A_t = \\alpha + \\beta\\, B_t + \\varepsilon_t$$\n",
        "\n",
        "The residual `ε_t = A_t − α − β B_t` is the **spread**. If A and B are cointegrated, this spread is stationary — it has a stable mean and reverts to it. We estimate β with `np.polyfit` (degree-1 least squares) and confirm the spread is mean-zero and well-behaved.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "fincept_title": "Hedge Ratio & Spread"
      },
      "outputs": [],
      "source": [
        "try:\n",
        "    import numpy as np\n",
        "    import pandas as pd\n",
        "except ImportError:\n",
        "    raise SystemExit(\"Fincept Notebook needs pandas + numpy — install with: pip install pandas numpy\")\n",
        "\n",
        "np.random.seed(7)\n",
        "\n",
        "N = 750\n",
        "factor = np.cumsum(np.random.normal(0.0, 1.0, N)) + 100.0\n",
        "noise_a = np.random.normal(0.0, 0.6, N)\n",
        "noise_b = np.random.normal(0.0, 0.6, N)\n",
        "A = 1.0 * factor + 20.0 + noise_a\n",
        "B = 0.6 * factor + 50.0 + noise_b\n",
        "\n",
        "# OLS regression of A on B: A = alpha + beta*B  (polyfit returns [slope, intercept])\n",
        "beta, alpha = np.polyfit(B, A, 1)\n",
        "\n",
        "spread = A - (alpha + beta * B)   # residual = stationary spread\n",
        "\n",
        "print(\"Least-squares hedge ratio (A regressed on B)\")\n",
        "print(f\"  beta  = {beta:.4f}   (true factor-loading ratio = 1.0/0.6 = {1.0/0.6:.4f})\")\n",
        "print(f\"  alpha = {alpha:.4f}\")\n",
        "print()\n",
        "print(\"Spread = A - (alpha + beta*B)\")\n",
        "print(f\"  mean = {spread.mean():.6f}  (should be ~0 by construction of OLS residual)\")\n",
        "print(f\"  std  = {spread.std():.4f}\")\n",
        "print(f\"  min  = {spread.min():.4f}   max = {spread.max():.4f}\")\n",
        "print()\n",
        "# Quick stationarity intuition: lag-1 autocorrelation well below 1 => mean-reverting\n",
        "s = pd.Series(spread)\n",
        "print(f\"  lag-1 autocorrelation of spread = {s.autocorr(1):.4f}  (<<1 => mean-reverting)\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 3. Rolling z-score of the spread\n",
        "\n",
        "To decide when the spread is \"too far\" from its mean we standardize it into a **z-score** using a *rolling* window (so the strategy only ever uses information available up to that day — no look-ahead):\n",
        "\n",
        "$$z_t = \\frac{\\text{spread}_t - \\text{mean}_{t-w..t}}{\\text{std}_{t-w..t}}$$\n",
        "\n",
        "A z-score of +2 means the spread sits two rolling standard deviations *above* normal — A is rich relative to B, a signal to short the spread. We use a 60-day window.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "fincept_title": "Rolling Z-Score"
      },
      "outputs": [],
      "source": [
        "try:\n",
        "    import numpy as np\n",
        "    import pandas as pd\n",
        "except ImportError:\n",
        "    raise SystemExit(\"Fincept Notebook needs pandas + numpy — install with: pip install pandas numpy\")\n",
        "\n",
        "np.random.seed(7)\n",
        "\n",
        "N = 750\n",
        "factor = np.cumsum(np.random.normal(0.0, 1.0, N)) + 100.0\n",
        "noise_a = np.random.normal(0.0, 0.6, N)\n",
        "noise_b = np.random.normal(0.0, 0.6, N)\n",
        "A = 1.0 * factor + 20.0 + noise_a\n",
        "B = 0.6 * factor + 50.0 + noise_b\n",
        "beta, alpha = np.polyfit(B, A, 1)\n",
        "spread = pd.Series(A - (alpha + beta * B))\n",
        "\n",
        "WIN = 60\n",
        "roll_mean = spread.rolling(WIN).mean()\n",
        "roll_std  = spread.rolling(WIN).std()\n",
        "zscore = (spread - roll_mean) / roll_std\n",
        "\n",
        "df = pd.DataFrame({\"spread\": spread, \"roll_mean\": roll_mean,\n",
        "                   \"roll_std\": roll_std, \"zscore\": zscore})\n",
        "\n",
        "valid = df.dropna()\n",
        "print(f\"Rolling z-score with {WIN}-day window\")\n",
        "print(f\"  valid (post-warmup) observations: {len(valid)} of {len(df)}\")\n",
        "print()\n",
        "print(\"Z-score distribution:\")\n",
        "print(valid[\"zscore\"].describe().round(4).to_string())\n",
        "print()\n",
        "print(f\"  days with z > +2 : {(valid['zscore'] >  2).sum()}\")\n",
        "print(f\"  days with z < -2 : {(valid['zscore'] < -2).sum()}\")\n",
        "print()\n",
        "print(\"Tail of the z-score frame:\")\n",
        "print(df.tail(8).round(4).to_string())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 4. Signals and backtest\n",
        "\n",
        "The trading rules, applied to the **spread position** (where +1 = long the spread = long A / short β·B, and −1 = short the spread):\n",
        "\n",
        "- **Enter short spread** (position = −1) when `z > +2` (spread too high, expect it to fall).\n",
        "- **Enter long spread** (position = +1) when `z < −2` (spread too low, expect it to rise).\n",
        "- **Exit to flat** (position = 0) when `|z| < 0.5` (reverted to the mean).\n",
        "- Otherwise **hold** the current position.\n",
        "\n",
        "We carry the position forward and earn the **change in spread × position held yesterday** (trade on tomorrow's move, not today's — no look-ahead). We then tally trades, hit rate, total return, and an annualized Sharpe of the daily strategy P&L.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "fincept_title": "Signals & P&L"
      },
      "outputs": [],
      "source": [
        "try:\n",
        "    import numpy as np\n",
        "    import pandas as pd\n",
        "except ImportError:\n",
        "    raise SystemExit(\"Fincept Notebook needs pandas + numpy — install with: pip install pandas numpy\")\n",
        "\n",
        "np.random.seed(7)\n",
        "\n",
        "N = 750\n",
        "factor = np.cumsum(np.random.normal(0.0, 1.0, N)) + 100.0\n",
        "noise_a = np.random.normal(0.0, 0.6, N)\n",
        "noise_b = np.random.normal(0.0, 0.6, N)\n",
        "A = 1.0 * factor + 20.0 + noise_a\n",
        "B = 0.6 * factor + 50.0 + noise_b\n",
        "beta, alpha = np.polyfit(B, A, 1)\n",
        "spread = pd.Series(A - (alpha + beta * B))\n",
        "\n",
        "WIN = 60\n",
        "ENTRY, EXIT = 2.0, 0.5\n",
        "z = (spread - spread.rolling(WIN).mean()) / spread.rolling(WIN).std()\n",
        "\n",
        "# Build positions with a stateful pass over the z-score\n",
        "pos = np.zeros(len(z))\n",
        "cur = 0\n",
        "for t in range(len(z)):\n",
        "    zt = z.iloc[t]\n",
        "    if np.isnan(zt):\n",
        "        cur = 0\n",
        "    elif cur == 0:\n",
        "        if zt > ENTRY:      cur = -1   # short the spread\n",
        "        elif zt < -ENTRY:   cur = +1   # long the spread\n",
        "    else:\n",
        "        if abs(zt) < EXIT:  cur = 0    # revert -> exit flat\n",
        "    pos[t] = cur\n",
        "position = pd.Series(pos, index=z.index)\n",
        "\n",
        "# P&L: yesterday's position earns today's spread change (no look-ahead)\n",
        "dspread = spread.diff()\n",
        "pnl = position.shift(1) * dspread\n",
        "pnl = pnl.fillna(0.0)\n",
        "\n",
        "# Trade accounting: a trade is a transition into a non-zero position\n",
        "entries = ((position != 0) & (position.shift(1).fillna(0) == 0))\n",
        "n_trades = int(entries.sum())\n",
        "\n",
        "# Per-trade P&L: sum pnl over each holding episode\n",
        "trade_id = entries.cumsum().where(position != 0)\n",
        "trade_pnl = pnl.groupby(trade_id).sum()\n",
        "wins = int((trade_pnl > 0).sum())\n",
        "hit_rate = wins / n_trades if n_trades else float(\"nan\")\n",
        "\n",
        "total_return = float(pnl.sum())\n",
        "daily = pnl[position.shift(1).fillna(0) != 0]   # P&L only while positioned\n",
        "sharpe = (daily.mean() / daily.std() * np.sqrt(252)) if daily.std() > 0 else float(\"nan\")\n",
        "\n",
        "print(\"=== Pairs strategy backtest ===\")\n",
        "metrics = pd.DataFrame({\n",
        "    \"metric\": [\"# trades\", \"# winning trades\", \"hit rate\",\n",
        "               \"total spread P&L\", \"avg P&L / trade\", \"annualized Sharpe\"],\n",
        "    \"value\":  [n_trades, wins, f\"{hit_rate:.2%}\",\n",
        "               f\"{total_return:.4f}\", f\"{trade_pnl.mean():.4f}\", f\"{sharpe:.4f}\"],\n",
        "})\n",
        "print(metrics.to_string(index=False))\n",
        "print()\n",
        "sig = pd.DataFrame({\"spread\": spread, \"zscore\": z,\n",
        "                    \"position\": position, \"pnl\": pnl})\n",
        "print(\"Tail of signals / P&L frame:\")\n",
        "print(sig.tail(10).round(4).to_string())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 5. Reading the results\n",
        "\n",
        "A working pairs strategy on a genuinely cointegrated pair should show a positive total spread P&L, a hit rate comfortably above 50%, and a Sharpe materially above zero — earned with low correlation to the broad market because every position is a hedged long/short. The z-score thresholds are the key tuning knobs: a wider entry band (e.g. ±2.5) trades less often but with higher conviction; a tighter exit band holds longer for fuller reversion.\n",
        "\n",
        "**Real-world cautions:** cointegration can *break* — if the structural link between the two names dissolves, the spread trends instead of reverting and the strategy bleeds. Production systems re-estimate β on a rolling window, monitor the spread with a formal stationarity test (ADF), add stop-losses, and account for transaction costs and borrow fees on the short leg. The z-score engine you built here is the core; risk management is what keeps it alive.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "---\n",
        "*— Fincept Notebook · part of Fincept Terminal. Edit any cell and press Ctrl+Enter to run.*\n"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}