Skip to content

Backtesting API

asrquant.backtest

Auditable vectorized portfolio backtesting.

BacktestResult dataclass

All outputs required for analysis, audit, visualization, and export.

Source code in src/asrquant/backtest.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@dataclass
class BacktestResult:
    """All outputs required for analysis, audit, visualization, and export."""

    prices: pd.DataFrame
    asset_returns: pd.DataFrame
    target_weights: pd.DataFrame
    effective_weights: pd.DataFrame
    gross_returns: pd.Series
    net_returns: pd.Series
    equity: pd.Series
    turnover: pd.Series
    costs: pd.Series
    cost_breakdown: pd.DataFrame
    spec: BacktestSpec
    metadata: dict[str, Any]

    @property
    def fingerprint(self) -> str:
        return str(self.metadata["experiment_fingerprint"])


    @property
    def trades(self) -> pd.DataFrame:
        """Return an auditable target-weight change ledger."""
        changes = self.effective_weights.diff().fillna(self.effective_weights)
        records = []
        for timestamp, row in changes.iterrows():
            for asset, change in row.items():
                if abs(change) > 1e-15:
                    records.append({
                        "timestamp": timestamp,
                        "asset": asset,
                        "weight_change": float(change),
                        "direction": "buy" if change > 0 else "sell",
                        "price": float(self.prices.loc[timestamp, asset]),
                        "target_weight": float(self.effective_weights.loc[timestamp, asset]),
                    })
        if not records:
            return pd.DataFrame(columns=["asset", "weight_change", "direction", "price", "target_weight"]).rename_axis("timestamp")
        return pd.DataFrame(records).set_index("timestamp")

    @property
    def metrics(self) -> pd.Series:
        return summary_metrics(
            self.net_returns,
            annualization=self.spec.annualization,
            risk_free_rate=self.spec.risk_free_rate,
            turnover=self.turnover,
        )

    @property
    def summary(self) -> pd.Series:
        """Canonical compact result summary."""
        return self.metrics

    def to_dict(self) -> dict[str, Any]:
        """Serialize the analytical summary and reproducibility metadata."""
        return {
            "result_type": "backtest",
            "summary": self.metrics.to_dict(),
            "fingerprint": self.fingerprint,
            "metadata": dict(self.metadata),
            "spec": self.spec.to_dict(),
        }

    def compare(self, benchmark_returns: pd.Series) -> pd.Series:
        return summary_metrics(
            self.net_returns,
            annualization=self.spec.annualization,
            risk_free_rate=self.spec.risk_free_rate,
            benchmark=benchmark_returns,
            turnover=self.turnover,
        )

    def to_frame(self) -> pd.DataFrame:
        return pd.concat(
            {
                "gross_return": self.gross_returns,
                "net_return": self.net_returns,
                "equity": self.equity,
                "turnover": self.turnover,
                "cost": self.costs,
            },
            axis=1,
        )

    def report(self, output: str | None = None, title: str | None = None) -> str:
        from .report import create_html_report

        return create_html_report(self, output=output, title=title)

    def plot(self, kind: str = "dashboard", **kwargs: Any):
        from .viz.performance import PerformanceVisualizer

        viz = PerformanceVisualizer()
        if kind == "dashboard":
            return viz.dashboard(self, **kwargs)
        if not hasattr(viz, kind):
            raise ValueError(f"unknown plot kind: {kind}")
        return getattr(viz, kind)(self, **kwargs)

trades property 

trades: DataFrame

Return an auditable target-weight change ledger.

summary property 

summary: Series

Canonical compact result summary.

to_dict 

to_dict() -> dict[str, Any]

Serialize the analytical summary and reproducibility metadata.

Source code in src/asrquant/backtest.py 
88
89
90
91
92
93
94
95
96
def to_dict(self) -> dict[str, Any]:
    """Serialize the analytical summary and reproducibility metadata."""
    return {
        "result_type": "backtest",
        "summary": self.metrics.to_dict(),
        "fingerprint": self.fingerprint,
        "metadata": dict(self.metadata),
        "spec": self.spec.to_dict(),
    }

run_backtest 

run_backtest(prices: Series | DataFrame, target_weights: Series | DataFrame, spec: BacktestSpec | None = None) -> BacktestResult

Run a deterministic weight-based backtest under an explicit contract.

Source code in src/asrquant/backtest.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def run_backtest(
    prices: pd.Series | pd.DataFrame,
    target_weights: pd.Series | pd.DataFrame,
    spec: BacktestSpec | None = None,
) -> BacktestResult:
    """Run a deterministic weight-based backtest under an explicit contract."""
    spec = spec or BacktestSpec()
    spec.validate()
    price_frame = clean_prices(prices, spec.missing_data)
    weights = align_like(target_weights, price_frame, fill_value=0.0)
    weights = _rebalance_targets(weights, spec.rebalance)
    weights = _enforce_constraints(weights, spec)
    effective = weights.shift(spec.execution_delay).fillna(0.0)

    returns = simple_returns(price_frame)
    gross_returns = (effective * returns).sum(axis=1)

    previous = effective.shift(1).fillna(0.0)
    traded = (effective - previous).abs()
    turnover = traded.sum(axis=1)
    linear_cost = turnover * spec.costs.linear_bps / 10_000.0
    impact_cost = spec.costs.impact_coefficient * np.power(turnover, spec.costs.impact_exponent)
    short_exposure = (-effective.clip(upper=0.0)).sum(axis=1)
    borrow_cost = short_exposure * spec.costs.borrow_bps_annual / 10_000.0 / spec.annualization
    total_cost = linear_cost + impact_cost + borrow_cost

    cash_weight = 1.0 - effective.sum(axis=1)
    cash_return = cash_weight * spec.risk_free_rate / spec.annualization
    net_returns = gross_returns + cash_return - total_cost
    equity = spec.initial_capital * (1.0 + net_returns).cumprod()

    contract_flags = {
        "same_bar_execution_risk": spec.execution_delay == 0,
        "costs_enabled": bool(spec.costs.linear_bps or spec.costs.impact_coefficient or spec.costs.borrow_bps_annual),
        "shorting_enabled": not spec.long_only,
    }
    experiment_payload = f"{data_fingerprint(price_frame)}:{spec.fingerprint()}"
    metadata = {
        "data_fingerprint": data_fingerprint(price_frame),
        "spec_fingerprint": spec.fingerprint(),
        "experiment_fingerprint": experiment_payload,
        "contract_flags": contract_flags,
        "n_observations": len(price_frame),
        "n_assets": price_frame.shape[1],
    }
    cost_breakdown = pd.DataFrame(
        {
            "linear": linear_cost,
            "impact": impact_cost,
            "borrow": borrow_cost,
            "total": total_cost,
        },
        index=price_frame.index,
    )
    return BacktestResult(
        prices=price_frame,
        asset_returns=returns,
        target_weights=weights,
        effective_weights=effective,
        gross_returns=gross_returns.rename("gross_return"),
        net_returns=net_returns.rename("net_return"),
        equity=equity.rename("equity"),
        turnover=turnover.rename("turnover"),
        costs=total_cost.rename("cost"),
        cost_breakdown=cost_breakdown,
        spec=spec,
        metadata=metadata,
    )

compare_backtests 

compare_backtests(results: dict[str, BacktestResult]) -> pd.DataFrame

Compare any number of backtests on a common metric table.

Source code in src/asrquant/backtest.py 
205
206
207
208
209
def compare_backtests(results: dict[str, BacktestResult]) -> pd.DataFrame:
    """Compare any number of backtests on a common metric table."""
    if not results:
        raise ValueError("at least one result is required")
    return pd.DataFrame({name: result.metrics for name, result in results.items()}).T