Function run_mid_frequency_pipeline

python/openquant/pipeline.py:11–41

Run an end-to-end AFML-style research pipeline. Returns nested dictionaries with stage outputs: events, signals, portfolio, risk, backtest, leakage_checks.

(
    timestamps: Sequence[str],
    close: Sequence[float],
    model_probabilities: Sequence[float],
    asset_prices: Sequence[Sequence[float]],
    model_sides: Sequence[float] | None = None,
    asset_names: Sequence[str] | None = None,
    cusum_threshold: float = 0.001,
    num_classes: int = 2,
    step_size: float = 0.1,
    risk_free_rate: float = 0.0,
    confidence_level: float = 0.05,
)

#: Trading days (sessions) a year: the annualisation factor for daily bars.
TRADING_DAYS_PER_YEAR = 252.0
#: Minutes in one trading session (a 6.5-hour US equity session, 09:30 to 16:00).
SESSION_MINUTES = 390.0
#: One-minute bars a year, ``TRADING_DAYS_PER_YEAR * SESSION_MINUTES`` = 98,280.
MINUTE_BARS_PER_YEAR = TRADING_DAYS_PER_YEAR * SESSION_MINUTES


def infer_periods_per_year(timestamps: Sequence[str]) -> float | None:
    """Bars a year implied by the spacing of ``timestamps``.

    Thin wrapper over ``openquant._core.pipeline.infer_periods_per_year``: the median gap
    between consecutive strictly increasing timestamps is mapped with the convention of
    252 sessions of 390 minutes a year. Intraday gaps give ``252 * 390 / gap_minutes``
    (one-minute bars: 98,280); gaps from 20 hours to under 4 days give 252 (daily bars);
    longer gaps give ``365.25 / gap_days``.

    Parameters
    ----------
    timestamps : Sequence[str]
        Bar timestamps as ``"%Y-%m-%d %H:%M:%S"`` strings, oldest first.

    Returns
    -------
    float or None
        Bars a year, or None when fewer than two timestamps strictly increase.
    """
    return _core.pipeline.infer_periods_per_year(list(timestamps))


def _resolve_periods_per_year(timestamps: Sequence[str], periods_per_year: float | None) -> float:
    if periods_per_year is not None:
        return float(periods_per_year)
    inferred = infer_periods_per_year(timestamps)
    return TRADING_DAYS_PER_YEAR if inferred is None else inferred