feat(portfolio): add expanding window and purging to walk_forward

eval_data/ohlcv_sample.csv

@@ -0,0 +1,6 @@
+date,symbol,open,high,low,close,volume
+2020-01-01,AAPL,75,76,74,75.5,1000000
+2020-01-02,AAPL,75.5,77,75,76.8,1200000
+2020-01-03,AAPL,76.8,78,76,77.5,1100000
+2020-01-04,AAPL,77.5,79,77,78.2,1300000
+2020-01-05,AAPL,78.2,80,78,79.5,1250000

examples/data_split.py

@@ -0,0 +1,21 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+def split_data(df, train_size=3, test_size=1):
+    splits = []
+    for start in range(0, len(df) - train_size - test_size + 1):
+        train = df.iloc[start:start + train_size]
+        test = df.iloc[start + train_size:start + train_size + test_size]
+        splits.append((train, test))
+    return splits
+
+splits = split_data(df)
+
+for i, (train, test) in enumerate(splits):
+    print(f"Split {i}")
+    print("Train:")
+    print(train[["date", "close"]])
+    print("Test:")
+    print(test[["date", "close"]])
+    print("-" * 20)

examples/leaky_strategy.py

@@ -0,0 +1,11 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+# Intentionally bad: uses tomorrow's close today.
+df["past_return"] = df["close"] / df["close"].shift(1)
+
+df["signal"] = df["past_return"] > 1
+df["strategy_return"] = df["signal"] * df["past_return"]
+
+print(df[["date", "symbol", "close", "past_return", "signal", "strategy_return"]])

examples/metrics_report.py

@@ -0,0 +1,28 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+df["past_return"] = df["close"] / df["close"].shift(1)
+df["signal"] = df["past_return"] > 1
+
+fee_rate = 0.001
+slippage_rate = 0.0005
+
+df["trade"] = df["signal"].astype(int).diff().abs().fillna(df["signal"].astype(int))
+df["gross_return"] = df["signal"] * df["past_return"]
+df["cost"] = df["trade"] * (fee_rate + slippage_rate)
+df["net_return"] = (df["gross_return"] - df["cost"]).fillna(0)
+
+total_return = df["net_return"].sum()
+num_trades = int(df["trade"].sum())
+max_drawdown = (df["net_return"].cummax() - df["net_return"]).max()
+sharpe = df["net_return"].mean() / df["net_return"].std() if df["net_return"].std() != 0 else 0
+
+metrics = {
+    "total_return": total_return,
+    "sharpe": sharpe,
+    "max_drawdown": max_drawdown,
+    "num_trades": num_trades
+}
+
+print(metrics)

examples/safe_optimizer.py

@@ -0,0 +1,43 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+df["past_return"] = df["close"] / df["close"].shift(1)
+
+def run_strategy(data, threshold):
+    data = data.copy()
+    data["signal"] = data["past_return"] > threshold
+    data["strategy_return"] = data["signal"] * data["past_return"]
+    return data["strategy_return"].fillna(0).sum()
+
+def split_data(df, train_size=3, test_size=1):
+    splits = []
+    for start in range(0, len(df) - train_size - test_size + 1):
+        train = df.iloc[start:start + train_size]
+        test = df.iloc[start + train_size:start + train_size + test_size]
+        splits.append((train, test))
+    return splits
+
+thresholds = [1.005, 1.01, 1.015]
+results = []
+
+for split_id, (train, test) in enumerate(split_data(df)):
+    train_scores = {}
+
+    for threshold in thresholds:
+        train_scores[threshold] = run_strategy(train, threshold)
+
+    best_threshold = max(train_scores, key=train_scores.get)
+
+    test_score = run_strategy(test, best_threshold)
+
+    results.append({
+        "split": split_id,
+        "best_threshold": best_threshold,
+        "train_score": train_scores[best_threshold],
+        "test_score": test_score
+    })
+
+results_df = pd.DataFrame(results)
+
+print(results_df)

examples/trading_costs.py

@@ -0,0 +1,16 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+df["past_return"] = df["close"] / df["close"].shift(1)
+df["signal"] = df["past_return"] > 1
+
+fee_rate = 0.001      # 0.1% fee
+slippage_rate = 0.0005  # 0.05% slippage
+
+df["trade"] = df["signal"].astype(int).diff().abs().fillna(df["signal"].astype(int))
+df["gross_return"] = df["signal"] * df["past_return"]
+df["cost"] = df["trade"] * (fee_rate + slippage_rate)
+df["net_return"] = df["gross_return"] - df["cost"]
+
+print(df[["date", "close", "signal", "trade", "gross_return", "cost", "net_return"]])

examples/walk_forward.py

@@ -0,0 +1,28 @@
+import pandas as pd
+
+df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+df["past_return"] = df["close"] / df["close"].shift(1)
+
+# parameters
+train_size = 3
+test_size = 1
+
+results = []
+
+for start in range(0, len(df) - train_size - test_size + 1):
+    train = df.iloc[start:start + train_size]
+    test = df.iloc[start + train_size:start + train_size + test_size]
+
+    # simple rule learned from train
+    threshold = train["past_return"].mean()
+
+    test = test.copy()
+    test["signal"] = test["past_return"] > threshold
+    test["strategy_return"] = test["signal"] * test["past_return"]
+
+    results.append(test)
+
+final = pd.concat(results)
+
+print(final[["date", "close", "past_return", "signal", "strategy_return"]])

tests/test_leakage_detection.py

@@ -0,0 +1,13 @@
+import pandas as pd
+
+def test_no_future_data_used():
+    df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+    # SAFE logic (past only)
+    df["past_return"] = df["close"] / df["close"].shift(1)
+
+    # Ensure first value is NaN (no future access)
+    assert pd.isna(df["past_return"].iloc[0])
+
+    # Ensure no use of future data
+    assert "future_return" not in df.columns

tests/test_metrics_report.py

@@ -0,0 +1,21 @@
+import pandas as pd
+
+def test_metrics_exist():
+    df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+    df["past_return"] = df["close"] / df["close"].shift(1)
+    df["signal"] = df["past_return"] > 1
+    df["trade"] = df["signal"].astype(int).diff().abs().fillna(df["signal"].astype(int))
+    df["net_return"] = df["past_return"].fillna(0)
+
+    metrics = {
+        "total_return": df["net_return"].sum(),
+        "sharpe": 0,
+        "max_drawdown": 0,
+        "num_trades": int(df["trade"].sum())
+    }
+
+    assert "total_return" in metrics
+    assert "sharpe" in metrics
+    assert "max_drawdown" in metrics
+    assert "num_trades" in metrics

tests/test_portfolio_walk_forward.py

@@ -0,0 +1,139 @@
+import pytest
+import pandas as pd
+import vectorbt as vbt
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+# ─── M3: Enhanced features tests ───────────────────────────────────────────
+
+def test_walk_forward_expanding_window():
+    """Expanding window should grow train window from index 0 each fold."""
+    close = pd.Series([1.0, 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8])
+    pf = vbt.Portfolio.from_holding(close)
+    result = pf.walk_forward(train_size=3, test_size=2, expanding=True)
+
+    assert isinstance(result, pd.DataFrame)
+    # In expanding mode, first fold: train=[0:3] (n=3), test=[3:5] (n=2)
+    # Second fold: train=[0:4] (n=4), test=[4:6] (n=2)
+    # Third fold: train=[0:5] (n=5), test=[5:7] (n=2)
+    assert len(result) >= 3  # at least 3 folds + 1 summary row
+    # The last row should be the summary
+    assert result.iloc[-1]["split"] == "summary"
+    # Expanding windows should have increasing n_train
+    n_trains = result[result["split"] != "summary"]["n_train"].tolist()
+    assert n_trains == sorted(n_trains), f"n_train not increasing: {n_trains}"
+    # All folds should have window_type == "expanding"
+    assert all(result[result["split"] != "summary"]["window_type"] == "expanding")
+
+
+def test_walk_forward_purging_gap():
+    """Purging should create a gap between train and test windows."""
+    close = pd.Series([1.0] * 20)
+    pf = vbt.Portfolio.from_holding(close)
+    result = pf.walk_forward(train_size=5, test_size=2, purging=3, step_size=3)
+
+    for _, row in result[result["split"] != "summary"].iterrows():
+        # After purging, train_end + purging < test_start
+        # In the rolling case with purging=3: gap_start = train_end - 3
+        # train_returns = [train_start:gap_start], test_returns = [train_end:test_end]
+        # So there are purging periods between train and test
+        train_end_ts = pd.Timestamp(row["train_end"])
+        test_start_ts = pd.Timestamp(row["test_start"])
+        gap_days = (test_start_ts - train_end_ts).days
+        assert gap_days >= 3, f"Purging gap should be >= 3 days, got {gap_days}"
+
+
+def test_walk_forward_rolling_vs_expanding_diff():
+    """Rolling and expanding windows should produce different train windows."""
+    close = pd.Series([1.0 + i * 0.01 for i in range(30)])
+    pf = vbt.Portfolio.from_holding(close)
+
+    rolling = pf.walk_forward(train_size=5, test_size=2, expanding=False, step_size=3)
+    expanding = pf.walk_forward(train_size=5, test_size=2, expanding=True, step_size=3)
+
+    # Rolling fold 1: train=[0:5], test=[5:7]
+    # Expanding fold 1: train=[0:5], test=[5:7]  (same as rolling first fold)
+    # Expanding fold 2: train=[0:7], test=[7:9]  (train is larger than rolling would give)
+    rolling_n_trains = rolling[rolling["split"] != "summary"]["n_train"].tolist()
+    expanding_n_trains = expanding[expanding["split"] != "summary"]["n_train"].tolist()
+
+    # Expanding n_trains should be strictly increasing
+    assert expanding_n_trains == sorted(expanding_n_trains)
+    # Rolling n_trains should all be equal (fixed window)
+    assert len(set(rolling_n_trains)) == 1
+
+
+def test_walk_forward_summary_row():
+    """Summary row should contain mean/std/min/max of test metrics."""
+    close = pd.Series([1.0, 1.2, 1.1, 1.3, 1.0, 1.4, 1.2, 1.5, 1.3])
+    pf = vbt.Portfolio.from_holding(close)
+    result = pf.walk_forward(train_size=2, test_size=1, step_size=1)
+
+    assert isinstance(result, pd.DataFrame)
+    summary_row = result.iloc[-1]
+    assert summary_row["split"] == "summary"
+    assert "test_metric" in summary_row.index
+    assert "test_metric_std" in summary_row.index
+    assert "test_metric_min" in summary_row.index
+    assert "test_metric_max" in summary_row.index
+    assert summary_row["test_metric_std"] >= 0  # std is non-negative
+
+
+def test_walk_forward_purging_negative_error():
+    """Negative purging should raise ValueError."""
+    close = pd.Series([1.0, 1.2, 1.1])
+    pf = vbt.Portfolio.from_holding(close)
+    with pytest.raises(ValueError, match="purging"):
+        pf.walk_forward(train_size=1, test_size=1, purging=-1)
+
+
+def test_walk_forward_returns_dataframe():
+    close = pd.Series([1, 2, 3, 4, 5])
+    pf = vbt.Portfolio.from_holding(close)
+
+    result = pf.walk_forward(train_size=2, test_size=1)
+
+    assert isinstance(result, pd.DataFrame)
+    assert "train_start" in result.columns
+    assert "test_start" in result.columns
+    assert "train_metric" in result.columns
+    assert "test_metric" in result.columns
+
+
+def test_walk_forward_no_overlap():
+    close = pd.Series([1, 2, 3, 4, 5])
+    pf = vbt.Portfolio.from_holding(close)
+
+    result = pf.walk_forward(train_size=2, test_size=1)
+
+    for _, row in result[result["split"] != "summary"].iterrows():
+        assert row["train_end"] < row["test_start"]
+
+
+def test_walk_forward_invalid_sizes():
+    close = pd.Series([1, 2, 3, 4, 5])
+    pf = vbt.Portfolio.from_holding(close)
+
+    with pytest.raises(ValueError):
+        pf.walk_forward(train_size=0, test_size=1)
+
+    with pytest.raises(ValueError):
+        pf.walk_forward(train_size=2, test_size=0)
+
+    with pytest.raises(ValueError):
+        pf.walk_forward(train_size=2, test_size=1, step_size=0)
+

tests/test_safe_optimizer.py

@@ -0,0 +1,17 @@
+import pandas as pd
+
+def test_optimizer_uses_train_before_test():
+    df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+    df["past_return"] = df["close"] / df["close"].shift(1)
+
+    train = df.iloc[:3]
+    test = df.iloc[3:4]
+
+    assert train.index.max() < test.index.min()
+
+def test_optimizer_does_not_use_future_return():
+    df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+    df["past_return"] = df["close"] / df["close"].shift(1)
+
+    assert "future_return" not in df.columns
+    assert pd.isna(df["past_return"].iloc[0])

tests/test_walk_forward.py

@@ -0,0 +1,17 @@
+import pandas as pd
+
+def test_walk_forward_no_leakage():
+    df = pd.read_csv("eval_data/ohlcv_sample.csv", parse_dates=["date"])
+
+    df["past_return"] = df["close"] / df["close"].shift(1)
+
+    train = df.iloc[:3]
+    test = df.iloc[3:4]
+
+    threshold = train["past_return"].mean()
+
+    test = test.copy()
+    test["signal"] = test["past_return"] > threshold
+
+    # ensure test does not use future data
+    assert test.index.min() > train.index.max()