ParallelLLC/algorithmic_trading

1"""Multi-asset engine, panel, cross-sectional strategies and their null."""2 3from __future__ import annotations4 5import numpy as np6import pandas as pd7import pytest8 9from algotrader.attribution import build_style_factors, factor_attribution10from algotrader.cross_sectional import XS_REGISTRY, get_xs_strategy, scores_to_weights11from algotrader.data import simulate_ohlcv12from algotrader.engine import run_backtest13from algotrader.panel import Panel, load_panel14from algotrader.portfolio import (15    normalise_weights,16    rebalance_schedule,17    run_portfolio_backtest,18)19from algotrader.types import CostModel20from algotrader.validation.cross_permutation import (21    cross_sectional_permutation_test,22    permute_within_dates,23)24 25XS_KEYS = sorted(XS_REGISTRY)26 27 28def make_panel(n_assets=8, n=1200, drift_sd=0.0, seed=1, common_vol=0.008, idio=0.012) -> Panel:29    """A synthetic universe. ``drift_sd`` controls genuine cross-sectional structure."""30    rng = np.random.default_rng(seed)31    index = pd.date_range("2016-01-01", periods=n, freq="B")32    drift = rng.normal(0, drift_sd, n_assets)33    returns = (34        rng.normal(0.0002, common_vol, n)[:, None]35        + rng.normal(0, idio, (n, n_assets))36        + drift[None, :]37    )38    close = 100 * np.exp(np.cumsum(returns, axis=0))39    columns = [f"A{i}" for i in range(n_assets)]40 41    def frame(values):42        return pd.DataFrame(values, index=index, columns=columns)43 44    return Panel(45        fields={46            "open": frame(close),47            "high": frame(close * 1.004),48            "low": frame(close * 0.996),49            "close": frame(close),50            "volume": frame(np.full_like(close, 1e6)),51        },52        sources={c: "synthetic" for c in columns},53    )54 55 56@pytest.fixture(scope="module")57def panel() -> Panel:58    return make_panel()59 60 61class TestPanel:62    def test_fields_are_aligned(self, panel):63        assert panel.shape == (1200, 8)64        for name in ("open", "high", "low", "close", "volume"):65            assert panel.fields[name].index.equals(panel.index)66            assert list(panel.fields[name].columns) == panel.symbols67 68    def test_misaligned_fields_are_rejected(self, panel):69        broken = dict(panel.fields)70        broken["high"] = broken["high"].iloc[:-5]71        with pytest.raises(ValueError, match="not aligned"):72            Panel(fields=broken)73 74    def test_missing_field_is_rejected(self, panel):75        with pytest.raises(ValueError, match="missing field"):76            Panel(fields={"close": panel.close})77 78    def test_missing_data_is_not_forward_filled(self):79        """Filling a gap invents liquidity that never existed."""80        df = simulate_ohlcv("AAPL", "2018-01-01", "2022-01-01")81        gapped = df.drop(df.index[100:140])82        built = Panel.from_frames({"AAPL": gapped, "MSFT": df})83        assert built.close["AAPL"].isna().sum() == 4084 85    def test_tradable_requires_two_consecutive_prices(self, panel):86        assert not panel.tradable().iloc[0].any()87        assert panel.tradable().iloc[1:].all().all()88 89    def test_returns_are_masked_where_untradable(self, panel):90        assert panel.returns().iloc[0].isna().all()91 92    def test_delisting_is_detected(self):93        df = simulate_ohlcv("AAPL", "2016-01-01", "2022-01-01")94        dead = df.iloc[: len(df) // 2]95        built = Panel.from_frames({"ALIVE": df, "DEAD": dead})96        report = built.survivorship()97        assert report.n_delisted == 198        assert "DEAD" in report.delisted_symbols99        assert not report.biased100 101    def test_all_survivors_is_flagged_as_biased(self, panel):102        report = panel.survivorship()103        assert report.survival_rate == 1.0104        assert report.biased105        assert "upper bound" in report.note106 107    def test_load_panel_skips_symbols_without_history(self):108        built = load_panel(["SPY", "AAPL"], "2018-01-01", "2022-01-01", source="synthetic")109        assert set(built.symbols) == {"SPY", "AAPL"}110        assert all(s == "synthetic" for s in built.sources.values())111 112    def test_select_and_slice(self, panel):113        subset = panel.select(["A0", "A1"])114        assert subset.symbols == ["A0", "A1"]115        sliced = panel.slice(panel.index[10], panel.index[50])116        assert len(sliced) == 41117 118 119class TestPortfolioEngine:120    def test_matches_single_asset_engine_exactly(self):121        """One column through the matrix engine must equal the fast path."""122        df = simulate_ohlcv("SPY", "2018-01-01", "2023-01-01")123        single = Panel.from_frames({"SPY": df})124        costs = CostModel(1, 2, 50)125        n = len(df)126        cases = {127            "buy_and_hold": np.ones(n),128            "flip": np.tile([1.0, -1.0], n // 2 + 1)[:n],129            "long_flat": np.tile([1.0, 0.0], n // 2 + 1)[:n],130            "fractional": np.full(n, 0.5),131        }132        for name, target in cases.items():133            portfolio = run_portfolio_backtest(134                single, pd.DataFrame({"SPY": target}, index=df.index), costs=costs135            )136            direct = run_backtest(df, pd.Series(target, index=df.index), costs=costs)137            np.testing.assert_allclose(138                portfolio.returns.to_numpy(), direct.returns.to_numpy(),139                atol=1e-12, err_msg=f"mismatch for {name}",140            )141 142    def test_holding_still_costs_nothing_but_drift_does(self, panel):143        """A full-notional long needs no rebalancing; a short does."""144        costs = CostModel(10, 10, 0)145        long_only = pd.DataFrame(1.0 / len(panel.symbols), index=panel.index, columns=panel.symbols)146        result = run_portfolio_backtest(panel, long_only, costs=costs, gross_leverage=1.0)147        # Equal-weight across many names still drifts apart, so turnover > 0...148        assert result.metrics["turnover_ann"] > 0149        # ...but a single full-notional position does not.150        one = pd.DataFrame(0.0, index=panel.index, columns=panel.symbols)151        one["A0"] = 1.0152        assert run_portfolio_backtest(panel, one, costs=costs).costs.sum() == pytest.approx(153            20 / 1e4, rel=1e-6154        )155 156    def test_rebalancing_less_often_lowers_turnover(self, panel):157        weights = get_xs_strategy("equal_weight").generate(panel)158        turnovers = []159        for frequency in ("D", "M", "Q"):160            result = run_portfolio_backtest(161                panel, weights, costs=CostModel(1, 2, 0),162                rebalance_on=rebalance_schedule(panel.index, frequency),163            )164            turnovers.append(result.metrics["turnover_ann"])165        assert turnovers[0] > turnovers[1] > turnovers[2]166 167    def test_gross_leverage_is_capped(self, panel):168        weights = pd.DataFrame(1.0, index=panel.index, columns=panel.symbols)169        result = run_portfolio_backtest(panel, weights, gross_leverage=1.0)170        assert result.weights.abs().sum(axis=1).max() <= 1.0 + 1e-9171 172    def test_leverage_is_scaled_down_never_up(self, panel):173        small = pd.DataFrame(0.01, index=panel.index, columns=panel.symbols)174        result = run_portfolio_backtest(panel, small, gross_leverage=1.0)175        assert result.weights.abs().sum(axis=1).max() < 0.2176 177    def test_per_name_cap_is_applied(self, panel):178        weights = pd.DataFrame(0.0, index=panel.index, columns=panel.symbols)179        weights["A0"] = 1.0180        result = run_portfolio_backtest(panel, weights, max_weight=0.1)181        assert result.weights.abs().max().max() <= 0.1 + 1e-9182 183    def test_shorts_are_blocked_when_disallowed(self, panel):184        weights = pd.DataFrame(-0.1, index=panel.index, columns=panel.symbols)185        result = run_portfolio_backtest(panel, weights, allow_short=False)186        assert (result.weights >= 0).all().all()187 188    def test_delisted_names_cannot_be_held(self):189        df = simulate_ohlcv("AAPL", "2016-01-01", "2022-01-01")190        built = Panel.from_frames({"ALIVE": df, "DEAD": df.iloc[: len(df) // 2]})191        weights = pd.DataFrame(0.5, index=built.index, columns=built.symbols)192        result = run_portfolio_backtest(built, weights)193        after_death = built.close["DEAD"].last_valid_index()194        assert result.held.loc[result.held.index > after_death, "DEAD"].abs().max() == 0.0195 196    def test_lag_zero_is_rejected(self, panel):197        with pytest.raises(ValueError, match="lag"):198            run_portfolio_backtest(panel, pd.DataFrame(0.1, index=panel.index, columns=panel.symbols), lag=0)199 200    def test_future_bars_cannot_change_past_equity(self, panel):201        weights = get_xs_strategy("xs_momentum").generate(panel)202        full = run_portfolio_backtest(panel, weights)203        cut = run_portfolio_backtest(204            panel.slice(panel.index[0], panel.index[799]), weights.iloc[:800]205        )206        np.testing.assert_allclose(207            full.equity.iloc[:800].to_numpy(), cut.equity.to_numpy(), rtol=1e-10208        )209 210    def test_attribution_sums_to_gross_return(self, panel):211        weights = get_xs_strategy("equal_weight").generate(panel)212        result = run_portfolio_backtest(panel, weights, costs=CostModel(0, 0, 0))213        assert result.attribution().sum() == pytest.approx(result.gross_returns.sum(), rel=1e-9)214 215    def test_portfolio_metrics_are_reported(self, panel):216        result = run_portfolio_backtest(panel, get_xs_strategy("xs_momentum").generate(panel))217        for key in ("gross_exposure", "net_exposure", "avg_positions", "concentration_hhi"):218            assert key in result.metrics219 220 221class TestCrossSectionalStrategies:222    @pytest.mark.parametrize("key", XS_KEYS)223    def test_weights_are_well_formed(self, panel, key):224        weights = get_xs_strategy(key).generate(panel)225        assert weights.shape == panel.shape226        assert weights.notna().all().all()227        assert weights.abs().sum(axis=1).max() <= 1.0 + 1e-9228 229    @pytest.mark.parametrize("key", XS_KEYS)230    def test_weights_are_causal(self, panel, key):231        cut = 700232        full = get_xs_strategy(key).generate(panel).iloc[:cut]233        partial = get_xs_strategy(key).generate(panel.slice(panel.index[0], panel.index[cut - 1]))234        pd.testing.assert_frame_equal(full, partial, rtol=1e-9)235 236    def test_long_short_books_are_dollar_neutral(self, panel):237        weights = get_xs_strategy("xs_momentum").generate(panel)238        active = weights[weights.abs().sum(axis=1) > 1e-9]239        assert active.sum(axis=1).abs().max() < 1e-9240 241    def test_long_only_uses_the_whole_book(self):242        scores = pd.DataFrame(np.arange(40).reshape(4, 10).astype(float))243        weights = scores_to_weights(scores, long_frac=0.3, long_only=True)244        assert weights.sum(axis=1).round(9).eq(1.0).all()245        assert (weights >= 0).all().all()246 247    def test_thin_cross_sections_are_skipped(self):248        scores = pd.DataFrame(np.random.default_rng(0).normal(size=(10, 3)))249        assert scores_to_weights(scores).abs().sum(axis=1).max() == 0.0250 251    def test_equal_weight_holds_everything(self, panel):252        weights = get_xs_strategy("equal_weight").generate(panel)253        assert (weights.iloc[-1] > 0).all()254 255    def test_unknown_strategy_names_alternatives(self):256        with pytest.raises(KeyError, match="Available"):257            get_xs_strategy("nope")258 259 260class TestCrossSectionalNull:261    def test_permutation_preserves_the_shape_of_the_book(self, panel):262        weights = get_xs_strategy("xs_momentum").generate(panel)263        investable = panel.close.notna()264        shuffled = permute_within_dates(weights, investable, np.random.default_rng(0))265 266        np.testing.assert_allclose(267            np.sort(weights.to_numpy(), axis=1), np.sort(shuffled.to_numpy(), axis=1), atol=1e-12268        )269        np.testing.assert_allclose(270            weights.abs().sum(axis=1).to_numpy(), shuffled.abs().sum(axis=1).to_numpy(), atol=1e-12271        )272        np.testing.assert_allclose(273            weights.sum(axis=1).to_numpy(), shuffled.sum(axis=1).to_numpy(), atol=1e-12274        )275 276    def test_permutation_actually_reassigns(self, panel):277        weights = get_xs_strategy("xs_momentum").generate(panel)278        shuffled = permute_within_dates(weights, panel.close.notna(), np.random.default_rng(0))279        assert not np.allclose(weights.to_numpy(), shuffled.to_numpy())280 281    def test_non_investable_names_stay_empty(self):282        df = simulate_ohlcv("AAPL", "2016-01-01", "2022-01-01")283        built = Panel.from_frames({"ALIVE": df, "DEAD": df.iloc[: len(df) // 2]})284        weights = pd.DataFrame(0.5, index=built.index, columns=built.symbols)285        shuffled = permute_within_dates(weights, built.close.notna(), np.random.default_rng(1))286        after_death = built.close["DEAD"].last_valid_index()287        assert shuffled.loc[shuffled.index > after_death, "DEAD"].abs().max() == 0.0288 289    def test_real_cross_sectional_skill_is_detected(self):290        strong = make_panel(drift_sd=0.003, seed=11)291        weights = get_xs_strategy("xs_momentum").generate(strong)292        result = cross_sectional_permutation_test(293            strong, weights, n_permutations=100,294            rebalance_on=rebalance_schedule(strong.index, "M"), seed=2,295        )296        assert result.observed > result.null_mean297        assert result.p_value < 0.05298 299    def test_no_structure_is_not_significant(self):300        flat = make_panel(drift_sd=0.0, seed=12)301        weights = get_xs_strategy("xs_momentum").generate(flat)302        result = cross_sectional_permutation_test(303            flat, weights, n_permutations=100,304            rebalance_on=rebalance_schedule(flat.index, "M"), seed=4,305        )306        assert result.p_value > 0.05307 308    def test_p_value_can_never_be_zero(self, panel):309        weights = get_xs_strategy("xs_momentum").generate(panel)310        result = cross_sectional_permutation_test(panel, weights, n_permutations=20, seed=0)311        assert result.p_value >= 1 / 21312 313 314class TestAttribution:315    def test_equal_weight_is_explained_by_the_market(self, panel):316        factors = build_style_factors(panel)317        weights = get_xs_strategy("equal_weight").generate(panel)318        result = run_portfolio_backtest(panel, weights)319        report = factor_attribution(result.returns, factors)320 321        assert report["available"]322        assert report["r_squared"] > 0.85323        assert report["dominant_factor"] == "market"324        assert not report["alpha_significant"]325        assert "more cheaply" in report["note"]326 327    def test_a_factor_regressed_on_itself_has_no_alpha(self, panel):328        factors = build_style_factors(panel)329        report = factor_attribution(factors["market"], factors)330        assert abs(report["alpha_annual"]) < 0.05331        assert report["r_squared"] > 0.99332 333    def test_pure_noise_has_no_alpha_and_no_fit(self, panel):334        rng = np.random.default_rng(5)335        noise = pd.Series(rng.normal(0, 0.01, len(panel)), index=panel.index)336        report = factor_attribution(noise, build_style_factors(panel))337        assert not report["alpha_significant"]338        assert report["r_squared"] < 0.2339 340    def test_short_series_degrade_gracefully(self, panel):341        factors = build_style_factors(panel)342        report = factor_attribution(factors["market"].iloc[:20], factors)343        assert not report["available"]344        assert report["note"]345 346 347class TestPortfolioLab:348    def test_full_pipeline(self):349        from algotrader.portfolio_lab import PortfolioLabConfig, run_portfolio_lab350 351        report = run_portfolio_lab(352            PortfolioLabConfig(353                symbols=["SPY", "QQQ", "AAPL", "MSFT", "NVDA", "GLD"],354                start="2017-01-01", end="2023-01-01", source="synthetic",355                strategy="xs_momentum", n_permutations=20, wf_folds=2, grid_limit=6,356            )357        )358        assert 0.0 <= report.verdict["score"] <= 100.0359        assert report.verdict["grade"] in {"A", "B", "C", "D", "F"}360        assert 0 < report.permutation.p_value <= 1361        assert report.attribution["available"]362        assert report.survivorship.n_symbols == 6363        assert report.cost_stress["sharpe_3x"] <= report.cost_stress["sharpe_1x"] + 1e-9364 365    def test_survivorship_bias_reaches_the_verdict(self):366        from algotrader.portfolio_lab import PortfolioLabConfig, run_portfolio_lab367 368        report = run_portfolio_lab(369            PortfolioLabConfig(370                symbols=["SPY", "QQQ", "AAPL", "MSFT", "NVDA", "GLD"],371                start="2015-01-01", end="2023-01-01", source="synthetic",372                strategy="equal_weight", n_permutations=0, wf_folds=2, grid_limit=3,373            )374        )375        assert report.survivorship.biased376        assert any("still trading" in f for f in report.verdict["flags"])377        assert report.verdict["score"] <= 60378