Backtest Overfitting Guard with Deflated Sharpe Ratio and PBO

id: backtest-overfitting-guard
namespace: company.research

inputs:
  - id: start_year
    type: INT
    displayName: First year of the backtest sample
    defaults: 2000

  - id: lookback_days
    type: ARRAY
    itemType: INT
    displayName: Momentum lookback windows (trading days)
    defaults: [ 21, 63, 126, 252 ]

  - id: holding_days
    type: ARRAY
    itemType: INT
    displayName: Holding periods between rebalances (trading days)
    defaults: [ 5, 21, 63 ]

  - id: top_n
    type: INT
    displayName: Industries held long and short
    defaults: 5

  - id: cscv_blocks
    type: INT
    displayName: Number of CSCV blocks (even number)
    defaults: 16

  - id: dsr_threshold
    type: FLOAT
    displayName: Minimum Deflated Sharpe Ratio to publish without review
    defaults: 0.95

  - id: pbo_threshold
    type: FLOAT
    displayName: Maximum Probability of Backtest Overfitting to publish without review
    defaults: 0.5

  - id: notify_slack
    type: BOOL
    displayName: Post the verdict to Slack
    defaults: false

tasks:
  - id: download_industry_returns
    type: io.kestra.plugin.core.http.Download
    description: Daily returns of the 49 Fama-French industry portfolios from
      Kenneth French's data library.
    uri: https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/49_Industry_Portfolios_daily_CSV.zip

  - id: run_strategy_grid
    type: io.kestra.plugin.scripts.python.Script
    description: Backtests every lookback x holding-period variant of a long-short
      industry momentum strategy.
    dependencies:
      - kestra
      - numpy
      - pandas
    inputFiles:
      industries.zip: "{{ outputs.download_industry_returns.uri }}"
    outputFiles:
      - trial_returns.csv
    script: |
      import io
      import zipfile

      import numpy as np
      import pandas as pd
      from kestra import Kestra

      lookbacks = {{ inputs.lookback_days | toJson }}
      holdings = {{ inputs.holding_days | toJson }}
      top_n = {{ inputs.top_n }}

      with zipfile.ZipFile("industries.zip") as archive:
          text = archive.read(archive.namelist()[0]).decode("latin-1")

      # The file holds several tables; keep the value-weighted daily one.
      lines = text.splitlines()
      start = next(i for i, line in enumerate(lines) if "Average Value Weighted Returns -- Daily" in line) + 1
      end = next(i for i in range(start + 1, len(lines)) if not lines[i].strip())
      returns = pd.read_csv(io.StringIO("\n".join(lines[start:end])), index_col=0)
      returns.index = pd.to_datetime(returns.index.astype(str), format="%Y%m%d")
      returns.columns = returns.columns.str.strip()
      returns = returns.where(returns > -99, np.nan) / 100
      returns = returns[returns.index.year >= {{ inputs.start_year }}].dropna(axis=1, how="any")

      # Rank on returns up to day t and trade from t+1, so no future data leaks into a signal.
      log_growth = np.log1p(returns).cumsum()
      trials = {}
      for lookback in lookbacks:
          for holding in holdings:
              weights = pd.DataFrame(0.0, index=returns.index, columns=returns.columns)
              for t in range(lookback, len(returns) - 1, holding):
                  ranked = (log_growth.iloc[t] - log_growth.iloc[t - lookback]).sort_values()
                  window = weights.index[t + 1:t + 1 + holding]
                  weights.loc[window, ranked.index[-top_n:]] = 1 / top_n
                  weights.loc[window, ranked.index[:top_n]] = -1 / top_n
              trials[f"L{lookback}_H{holding}"] = (weights * returns).sum(axis=1)

      trial_returns = pd.DataFrame(trials).iloc[max(lookbacks) + 1:]
      trial_returns.index.name = "date"
      trial_returns.to_csv("trial_returns.csv", float_format="%.8f")

      Kestra.outputs({
          "industries": int(returns.shape[1]),
          "trading_days": len(trial_returns),
          "first_day": str(trial_returns.index[0].date()),
          "last_day": str(trial_returns.index[-1].date()),
          "trials": len(trials),
      })

  - id: evaluate
    type: io.kestra.plugin.core.flow.Parallel
    description: The leaderboard and the overfitting tests read the same trial
      returns, so they run side by side.
    tasks:
      - id: leaderboard
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: Annualized return, volatility, Sharpe ratio and maximum drawdown
          per strategy variant.
        inputFiles:
          trial_returns.csv: "{{ outputs.run_strategy_grid.outputFiles['trial_returns.csv'] }}"
        outputFiles:
          - leaderboard
        sql: |
          CREATE TABLE daily AS
          UNPIVOT (SELECT * FROM read_csv_auto('trial_returns.csv'))
          ON COLUMNS(* EXCLUDE (date)) INTO NAME trial VALUE ret;

          CREATE TABLE equity AS
          SELECT trial, date, SUM(LN(1 + ret)) OVER (PARTITION BY trial ORDER BY date) AS log_equity
          FROM daily;

          CREATE TABLE drawdowns AS
          SELECT trial, MIN(EXP(log_equity - running_peak) - 1) AS max_drawdown
          FROM (
            SELECT trial, log_equity, MAX(log_equity) OVER (PARTITION BY trial ORDER BY date) AS running_peak
            FROM equity
          )
          GROUP BY trial;

          COPY (
            SELECT
              d.trial,
              ROUND(AVG(d.ret) * 252, 4) AS annual_return,
              ROUND(STDDEV_SAMP(d.ret) * SQRT(252), 4) AS annual_volatility,
              ROUND(AVG(d.ret) / STDDEV_SAMP(d.ret) * SQRT(252), 3) AS sharpe_ratio,
              ROUND(ANY_VALUE(dd.max_drawdown), 4) AS max_drawdown
            FROM daily d
            JOIN drawdowns dd USING (trial)
            GROUP BY d.trial
            ORDER BY sharpe_ratio DESC
          ) TO '{{ outputFiles.leaderboard }}' (HEADER, DELIMITER ',');

      - id: overfitting_tests
        type: io.kestra.plugin.scripts.python.Script
        description: Deflated Sharpe Ratio (Bailey and Lopez de Prado, 2014) and
          Probability of Backtest Overfitting via CSCV (Bailey et al., 2017).
        dependencies:
          - kestra
          - numpy
          - pandas
        inputFiles:
          trial_returns.csv: "{{ outputs.run_strategy_grid.outputFiles['trial_returns.csv'] }}"
        script: |
          import itertools
          import math
          from statistics import NormalDist

          import numpy as np
          import pandas as pd
          from kestra import Kestra

          trial_returns = pd.read_csv("trial_returns.csv", index_col="date")
          returns = trial_returns.to_numpy()
          n_obs, n_trials = returns.shape
          blocks = {{ inputs.cscv_blocks }}
          norm = NormalDist()

          def sharpe(sample):
              std = sample.std(axis=0, ddof=1)
              return np.divide(sample.mean(axis=0), std, out=np.zeros_like(std), where=std > 0)

          # Deflated Sharpe Ratio on per-period Sharpe ratios.
          trial_sharpes = sharpe(returns)
          best = int(np.argmax(trial_sharpes))
          best_sharpe = trial_sharpes[best]
          centered = returns[:, best] - returns[:, best].mean()
          skew = (centered ** 3).mean() / centered.std() ** 3
          kurtosis = (centered ** 4).mean() / centered.std() ** 4
          euler_gamma = 0.5772156649
          expected_max_sharpe = math.sqrt(trial_sharpes.var(ddof=1)) * (
              (1 - euler_gamma) * norm.inv_cdf(1 - 1 / n_trials)
              + euler_gamma * norm.inv_cdf(1 - 1 / (n_trials * math.e))
          )
          dsr = norm.cdf(
              (best_sharpe - expected_max_sharpe) * math.sqrt(n_obs - 1)
              / math.sqrt(1 - skew * best_sharpe + (kurtosis - 1) / 4 * best_sharpe ** 2)
          )

          # Probability of Backtest Overfitting: how often the in-sample winner ranks below median out of sample.
          block_rows = np.array_split(np.arange(n_obs), blocks)
          logits = []
          for in_sample in itertools.combinations(range(blocks), blocks // 2):
              is_rows = np.concatenate([block_rows[b] for b in in_sample])
              oos_rows = np.concatenate([block_rows[b] for b in range(blocks) if b not in in_sample])
              chosen = int(np.argmax(sharpe(returns[is_rows])))
              oos_sharpes = sharpe(returns[oos_rows])
              omega = ((oos_sharpes < oos_sharpes[chosen]).sum() + 1) / (n_trials + 1)
              logits.append(math.log(omega / (1 - omega)))
          pbo = float(np.mean(np.array(logits) <= 0))

          Kestra.outputs({
              "best_trial": trial_returns.columns[best],
              "best_sharpe_annualized": round(float(best_sharpe * math.sqrt(252)), 3),
              "expected_max_sharpe_annualized": round(float(expected_max_sharpe * math.sqrt(252)), 3),
              "deflated_sharpe_ratio": round(float(dsr), 4),
              "pbo": round(pbo, 4),
              "cscv_combinations": len(logits),
          })

  - id: overfitting_gate
    type: io.kestra.plugin.core.flow.If
    description: A result that is not statistically distinguishable from luck waits
      for a researcher instead of being published.
    condition: "{{ outputs.overfitting_tests.vars.deflated_sharpe_ratio <
      inputs.dsr_threshold or outputs.overfitting_tests.vars.pbo >
      inputs.pbo_threshold }}"
    then:
      - id: flag_for_review
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: >-
          {{ outputs.overfitting_tests.vars.best_trial }} is likely overfit: DSR
          {{ outputs.overfitting_tests.vars.deflated_sharpe_ratio }} (threshold
          {{ inputs.dsr_threshold }}), PBO {{ outputs.overfitting_tests.vars.pbo
          }} (threshold {{ inputs.pbo_threshold }}). Resume this execution to
          publish the report anyway, or kill it to discard the result.

      - id: researcher_review
        type: io.kestra.plugin.core.flow.Pause
        description: Waits up to seven days for a researcher to resume or kill the
          execution.
        pauseDuration: P7D
        behavior: FAIL
    else:
      - id: passed_checks
        type: io.kestra.plugin.core.log.Log
        message: "{{ outputs.overfitting_tests.vars.best_trial }} passed both
          overfitting checks."

  - id: build_report
    type: io.kestra.plugin.scripts.python.Script
    description: Writes a Markdown research note with the verdict and the full leaderboard.
    dependencies:
      - kestra
    inputFiles:
      leaderboard.csv: "{{ outputs.leaderboard.outputFiles.leaderboard }}"
    outputFiles:
      - report.md
    script: |
      import csv

      grid = {{ outputs.run_strategy_grid.vars | toJson }}
      tests = {{ outputs.overfitting_tests.vars | toJson }}
      pbo_threshold = {{ inputs.pbo_threshold }}
      flagged = tests["deflated_sharpe_ratio"] < {{ inputs.dsr_threshold }} or tests["pbo"] > pbo_threshold

      with open("leaderboard.csv") as f:
          rows = list(csv.DictReader(f))

      table = ["| Variant | Annual return | Volatility | Sharpe | Max drawdown |", "|---|---|---|---|---|"]
      table += [
          f"| {r['trial']} | {float(r['annual_return']):.2%} | {float(r['annual_volatility']):.2%} | {r['sharpe_ratio']} | {float(r['max_drawdown']):.2%} |"
          for r in rows
      ]

      verdict = "Likely overfit, published after researcher review" if flagged else "Passed both overfitting checks"
      report = f"""# Industry momentum: backtest overfitting report

      **Verdict:** {verdict}

      Sample: {grid['industries']} industries, {grid['trading_days']} trading days ({grid['first_day']} to {grid['last_day']}), {grid['trials']} strategy variants.

      | Check | Value | Threshold |
      |---|---|---|
      | Best variant | {tests['best_trial']} | |
      | Best annualized Sharpe | {tests['best_sharpe_annualized']} | |
      | Expected best Sharpe from {grid['trials']} unskilled variants | {tests['expected_max_sharpe_annualized']} | |
      | Deflated Sharpe Ratio | {tests['deflated_sharpe_ratio']} | >= {{ inputs.dsr_threshold }} |
      | Probability of Backtest Overfitting | {tests['pbo']:.1%} | <= {pbo_threshold:.0%} |

      ## Leaderboard

      """ + "\n".join(table) + "\n"

      with open("report.md", "w") as f:
          f.write(report)

  - id: notify
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.notify_slack }}"
    then:
      - id: slack_verdict
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": {{ ('Backtest overfitting report: best variant ' ~ outputs.overfitting_tests.vars.best_trial ~ ', DSR ' ~ outputs.overfitting_tests.vars.deflated_sharpe_ratio ~ ', PBO ' ~ outputs.overfitting_tests.vars.pbo ~ '. Execution ' ~ execution.id) | toJson }}
          }

triggers:
  - id: weekly_after_data_refresh
    type: io.kestra.plugin.core.trigger.Schedule
    description: Monday 06:00 UTC. Shipped disabled; enable it once the defaults
      suit your research.
    cron: "0 6 * * 1"
    disabled: true