Signal API

MultiFrequencySignal

mfdro.signal.MultiFrequencySignal(config: SignalConfig | None = None)

Estimate a reproducible geometric disagreement signal.

Create an estimator from a validated scientific configuration.

Source code in src/mfdro/signal.py

def __init__(self, config: SignalConfig | None = None):
    """Create an estimator from a validated scientific configuration."""

    if config is not None and not isinstance(config, SignalConfig):
        raise TypeError("config must be a SignalConfig instance or None.")
    self.config = SignalConfig.reference() if config is None else config

estimate(measures: Mapping[str, object], *, seed: int | None = None, include_support: bool = False) -> SignalEstimate

Estimate one aligned collection of empirical measures.

DataFrames are aligned by asset label; unlabelled arrays must already share column order. seed controls projected directions, while include_support returns a defensive copy of a free-support center.

Source code in src/mfdro/signal.py

def estimate(
    self,
    measures: Mapping[str, object],
    *,
    seed: int | None = None,
    include_support: bool = False,
) -> SignalEstimate:
    """Estimate one aligned collection of empirical measures.

    DataFrames are aligned by asset label; unlabelled arrays must already
    share column order. ``seed`` controls projected directions, while
    ``include_support`` returns a defensive copy of a free-support center.
    """

    if not isinstance(include_support, bool):
        raise TypeError("include_support must be a boolean.")
    prepared = prepare_measures(measures, self.config.frequencies)
    arrays = scale_measures(prepared.arrays, self.config)
    weights = dispersion_weights(arrays, self.config)
    center_weights = barycenter_weights(self.config)
    effective_seed = self.config.random_state if seed is None else validate_seed(seed, "seed")

    support: FloatArray | None = None
    if self.config.barycenter == "projected_quantile":
        rho, squared_distances = projected_quantile_dispersion_with_components(
            arrays,
            weights,
            center_weights,
            self.config,
            effective_seed,
        )
    else:
        support = free_support_barycenter(arrays, self.config, center_weights)
        if self.config.distance == "sliced":
            rho, squared_distances = sliced_dispersion_with_components(
                arrays,
                support,
                weights,
                self.config,
                effective_seed,
            )
        else:
            rho, squared_distances = exact_dispersion_with_components(
                arrays,
                support,
                weights,
            )

    if not math.isfinite(rho) or rho < 0:
        raise RuntimeError("Signal estimation did not produce a finite non-negative value.")
    return SignalEstimate(
        rho=float(rho),
        sqrt_rho=float(math.sqrt(rho)),
        seed=effective_seed,
        config_digest=self.config.digest,
        frequencies=self.config.frequencies,
        frequency_weights=tuple(float(value) for value in weights),
        frequency_squared_distances=tuple(float(value) for value in squared_distances),
        barycenter_weights=tuple(float(value) for value in center_weights),
        sample_sizes=prepared.sample_sizes,
        n_assets=prepared.n_assets,
        asset_labels=prepared.asset_labels,
        support=support.copy() if include_support and support is not None else None,
    )

validate_path_inputs(daily_returns: pd.DataFrame, *, lookback_months: int, formation_dates: Sequence[object] | None = None, memberships: Mapping[object, Sequence[object]] | None = None, reference_calendar: Sequence[object] | None = None, seed_namespace: str = 'signal') -> PathDiagnostics

Inspect every requested window without computing transport geometry.

Hard contract violations still raise :class:DataContractError. Ordinary warm-up, calendar, and frequency-sample insufficiencies are returned as diagnostic rows so users can inspect the complete schedule.

Source code in src/mfdro/signal.py

def validate_path_inputs(
    self,
    daily_returns: pd.DataFrame,
    *,
    lookback_months: int,
    formation_dates: Sequence[object] | None = None,
    memberships: Mapping[object, Sequence[object]] | None = None,
    reference_calendar: Sequence[object] | None = None,
    seed_namespace: str = "signal",
) -> PathDiagnostics:
    """Inspect every requested window without computing transport geometry.

    Hard contract violations still raise :class:`DataContractError`.
    Ordinary warm-up, calendar, and frequency-sample insufficiencies are
    returned as diagnostic rows so users can inspect the complete schedule.
    """

    context = _prepare_path_context(
        self.config,
        daily_returns,
        lookback_months=lookback_months,
        formation_dates=formation_dates,
        memberships=memberships,
        reference_calendar=reference_calendar,
        seed_namespace=seed_namespace,
    )
    rows: list[dict[str, object]] = []
    for formation_date in context.dates:
        outcome = _prepare_window(self.config, context, formation_date)
        row: dict[str, object] = {
            "date": outcome.formation_date,
            "formation_month": outcome.formation_month.to_timestamp("M"),
            "start_date": outcome.start_date,
            "n_assets": (
                len(outcome.assets)
                if isinstance(outcome, _PreparedWindow)
                else outcome.n_assets
            ),
        }
        if isinstance(outcome, _PreparedWindow):
            row.update({"status": "ready", "reason": None, "detail": None})
            sample_sizes = {name: len(frame) for name, frame in outcome.measures.items()}
        else:
            row.update(
                {
                    "status": "insufficient",
                    "reason": outcome.reason.value,
                    "detail": outcome.detail,
                }
            )
            sample_sizes = outcome.sample_sizes
        for name in self.config.frequencies:
            row[f"n_{name}"] = sample_sizes.get(name)
        rows.append(row)

    formations = pd.DataFrame(rows, columns=_diagnostic_columns(self.config.frequencies))
    return PathDiagnostics(
        formations=formations,
        config=self.config,
        source_start=context.source_index.min(),
        source_end=context.source_index.max(),
        n_observations=len(context.source),
        n_assets=len(context.source.columns),
    )

estimate_path(daily_returns: pd.DataFrame, *, lookback_months: int, formation_dates: Sequence[object] | None = None, memberships: Mapping[object, Sequence[object]] | None = None, reference_calendar: Sequence[object] | None = None, on_insufficient: Literal['skip', 'raise'] = 'skip', seed_namespace: str = 'signal', progress_callback: ProgressCallback | None = None) -> SignalPath

Estimate a monthly point-in-time signal path from a daily panel.

progress_callback receives one immutable :class:PathProgress notification after every estimated or skipped formation. No progress dependency or terminal output is imposed by the package.

Source code in src/mfdro/signal.py

def estimate_path(
    self,
    daily_returns: pd.DataFrame,
    *,
    lookback_months: int,
    formation_dates: Sequence[object] | None = None,
    memberships: Mapping[object, Sequence[object]] | None = None,
    reference_calendar: Sequence[object] | None = None,
    on_insufficient: Literal["skip", "raise"] = "skip",
    seed_namespace: str = "signal",
    progress_callback: ProgressCallback | None = None,
) -> SignalPath:
    """Estimate a monthly point-in-time signal path from a daily panel.

    ``progress_callback`` receives one immutable :class:`PathProgress`
    notification after every estimated or skipped formation. No progress
    dependency or terminal output is imposed by the package.
    """

    if on_insufficient not in {"skip", "raise"}:
        raise ValueError("on_insufficient must be 'skip' or 'raise'.")
    if progress_callback is not None and not callable(progress_callback):
        raise TypeError("progress_callback must be callable or None.")
    context = _prepare_path_context(
        self.config,
        daily_returns,
        lookback_months=lookback_months,
        formation_dates=formation_dates,
        memberships=memberships,
        reference_calendar=reference_calendar,
        seed_namespace=seed_namespace,
    )
    estimate_rows: list[dict[str, object]] = []
    audit_rows: list[dict[str, object]] = []
    skipped_rows: list[dict[str, object]] = []
    total = len(context.dates)

    for completed, formation_date in enumerate(context.dates, start=1):
        outcome = _prepare_window(self.config, context, formation_date)
        if isinstance(outcome, _InsufficientWindow):
            if on_insufficient == "raise":
                raise DataContractError(outcome.detail)
            skipped_rows.append(
                outcome.to_skipped_record(self.config.digest, context.lookback_months)
            )
            _notify_progress(
                progress_callback,
                PathProgress(
                    completed=completed,
                    total=total,
                    date=formation_date,
                    status="skipped",
                    reason=outcome.reason,
                ),
            )
            continue

        seed = stable_seed(
            self.config.random_state,
            context.seed_namespace,
            str(outcome.formation_month),
        )
        estimate = self.estimate(outcome.measures, seed=seed)
        estimate_rows.append(
            {
                "date": formation_date,
                "formation_month": outcome.formation_month.to_timestamp("M"),
                **estimate.to_record(),
            }
        )
        window_index = cast(pd.DatetimeIndex, outcome.window.index)
        audit_record: dict[str, object] = {
            "date": formation_date,
            "formation_month": outcome.formation_month.to_timestamp("M"),
            "start_date": outcome.start_date,
            "window_end": window_index.max(),
            "lookback_months": context.lookback_months,
            "n_assets": len(outcome.assets),
            "asset_order_digest": _asset_order_digest(outcome.assets),
            "no_future_observations": bool(window_index.max() <= formation_date),
            "matrix_is_full": bool(outcome.window.notna().all().all()),
            "config_digest": self.config.digest,
            "seed": seed,
        }
        for name, frame in outcome.measures.items():
            audit_record[f"n_{name}"] = len(frame)
        audit_rows.append(audit_record)
        _notify_progress(
            progress_callback,
            PathProgress(
                completed=completed,
                total=total,
                date=formation_date,
                status="estimated",
            ),
        )

    estimates = pd.DataFrame(
        estimate_rows,
        columns=_estimate_columns(self.config.frequencies),
    )
    audit = pd.DataFrame(
        audit_rows,
        columns=_audit_columns(self.config.frequencies),
    )
    skipped = pd.DataFrame(skipped_rows, columns=_skipped_columns())
    if not estimates.empty:
        estimates = estimates.sort_values("date", kind="stable").reset_index(drop=True)
        audit = audit.sort_values("date", kind="stable").reset_index(drop=True)
    if not skipped.empty:
        skipped = skipped.sort_values("date", kind="stable").reset_index(drop=True)
    return SignalPath(
        estimates=estimates,
        audit=audit,
        config=self.config,
        skipped=skipped,
    )

SignalEstimate

mfdro.signal.SignalEstimate(rho: float, sqrt_rho: float, seed: int, config_digest: str, frequencies: tuple[str, ...], frequency_weights: tuple[float, ...], barycenter_weights: tuple[float, ...], sample_sizes: tuple[int, ...], n_assets: int, asset_labels: tuple[object, ...] | None, support: FloatArray | None = None, frequency_squared_distances: tuple[float, ...] = ()) dataclass

Auditable result of one multi-frequency signal estimate.

rho is the configured squared dispersion and sqrt_rho its square root. frequency_squared_distances retains the unweighted distance from each frequency to the configured center. The remaining fields identify the numerical experiment. support is returned only when requested for a free-support estimate.

to_record() -> dict[str, object]

Flatten the estimate into a machine-readable record.

Source code in src/mfdro/signal.py

def to_record(self) -> dict[str, object]:
    """Flatten the estimate into a machine-readable record."""

    record: dict[str, object] = {
        "rho": self.rho,
        "sqrt_rho": self.sqrt_rho,
        "seed": self.seed,
        "config_digest": self.config_digest,
        "n_assets": self.n_assets,
    }
    for index, frequency in enumerate(self.frequencies):
        if self.frequency_squared_distances:
            record[f"distance2_{frequency}"] = self.frequency_squared_distances[index]
        record[f"lambda_{frequency}"] = self.frequency_weights[index]
        record[f"barycenter_lambda_{frequency}"] = self.barycenter_weights[index]
        record[f"n_{frequency}"] = self.sample_sizes[index]
    return record

to_series() -> pd.Series

Return the flattened estimate as a labelled pandas Series.

Source code in src/mfdro/signal.py

def to_series(self) -> pd.Series:
    """Return the flattened estimate as a labelled pandas Series."""

    return pd.Series(self.to_record(), name="signal_estimate")

PathDiagnostics

mfdro.signal.PathDiagnostics(formations: pd.DataFrame, config: SignalConfig, source_start: pd.Timestamp, source_end: pd.Timestamp, n_observations: int, n_assets: int) dataclass

Non-numerical readiness report for a requested walk-forward path.

n_formations: int property

Return the number of inspected formations.

n_ready: int property

Return the number of formations ready for numerical estimation.

n_insufficient: int property

Return the number of insufficient formations.

is_usable: bool property

Return whether at least one formation can be estimated.

summary() -> dict[str, object]

Return a compact machine-readable diagnostic summary.

Source code in src/mfdro/signal.py

def summary(self) -> dict[str, object]:
    """Return a compact machine-readable diagnostic summary."""

    return {
        "source_start": self.source_start,
        "source_end": self.source_end,
        "n_observations": self.n_observations,
        "n_assets": self.n_assets,
        "n_formations": self.n_formations,
        "n_ready": self.n_ready,
        "n_insufficient": self.n_insufficient,
        "is_usable": self.is_usable,
        "config_digest": self.config.digest,
    }

PathProgress

mfdro.signal.PathProgress(completed: int, total: int, date: pd.Timestamp, status: Literal['estimated', 'skipped'], reason: SkipReason | None = None) dataclass

One progress notification emitted by :meth:estimate_path.

SkipReason

mfdro.signal.SkipReason

Bases: str, Enum

Machine-readable reason why a requested formation was not estimated.

SignalPath

mfdro.signal.SignalPath(estimates: pd.DataFrame, audit: pd.DataFrame, config: SignalConfig, skipped: pd.DataFrame = pd.DataFrame()) dataclass

Signal estimates and window-level audit produced walk-forward.

The three DataFrames retain stable columns even when empty. Convenience accessors expose the most common series without discarding the complete audit tables.

rho: pd.Series property

Return a copy of rho indexed by successful formation date.

sqrt_rho: pd.Series property

Return a copy of sqrt_rho indexed by successful formation date.

successful_dates: pd.DatetimeIndex property

Return successful formation dates in path order.

skipped_dates: pd.DatetimeIndex property

Return skipped formation dates in path order.

dispersion_weights: pd.DataFrame property

Return dates and normalized dispersion weights.

center_weights: pd.DataFrame property

Return dates and normalized barycenter weights.

summary() -> dict[str, object]

Return a compact path summary without reducing the audit trail.

Source code in src/mfdro/signal.py

def summary(self) -> dict[str, object]:
    """Return a compact path summary without reducing the audit trail."""

    return {
        "n_estimates": len(self.estimates),
        "n_skipped": len(self.skipped),
        "first_estimate": (None if self.estimates.empty else self.estimates["date"].iloc[0]),
        "last_estimate": (None if self.estimates.empty else self.estimates["date"].iloc[-1]),
        "frequencies": self.config.frequencies,
        "config_digest": self.config.digest,
    }

save(directory: str | Path, *, overwrite: bool = False) -> Path

Persist results, audit, skipped formations, config, and checksums.

The portable JSON-table format avoids unsafe pickle deserialization and does not require an optional Parquet engine.

Source code in src/mfdro/signal.py

def save(self, directory: str | Path, *, overwrite: bool = False) -> Path:
    """Persist results, audit, skipped formations, config, and checksums.

    The portable JSON-table format avoids unsafe pickle deserialization and
    does not require an optional Parquet engine.
    """

    if not isinstance(overwrite, bool):
        raise TypeError("overwrite must be a boolean.")
    destination = Path(directory).expanduser().resolve()
    if destination.exists() and not destination.is_dir():
        raise FileExistsError(f"Signal path destination is not a directory: {destination}")
    if destination.exists() and any(destination.iterdir()) and not overwrite:
        raise FileExistsError(
            f"Signal path destination is not empty: {destination}. "
            "Pass overwrite=True to replace MFDRO files."
        )
    destination.mkdir(parents=True, exist_ok=True)

    payloads = {
        "config.json": self.config.to_json(),
        "estimates.json": _frame_to_json(self.estimates),
        "audit.json": _frame_to_json(self.audit),
        "skipped.json": _frame_to_json(self.skipped),
    }
    for name, payload in payloads.items():
        destination.joinpath(name).write_text(payload, encoding="utf-8")
    manifest = {
        "format_version": PATH_FORMAT_VERSION,
        "package_version": _package_version(),
        "config_digest": self.config.digest,
        "rows": {
            "estimates": len(self.estimates),
            "audit": len(self.audit),
            "skipped": len(self.skipped),
        },
        "sha256": {
            name: hashlib.sha256(payload.encode("utf-8")).hexdigest()
            for name, payload in payloads.items()
        },
    }
    destination.joinpath("manifest.json").write_text(
        json.dumps(manifest, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    return destination

load(directory: str | Path) -> SignalPath classmethod

Load and integrity-check a path written by :meth:save.

Format-1 paths remain readable. Their per-frequency distances cannot be reconstructed from the aggregate signal, so the migrated columns contain NaN. Saving the returned object writes the current format.

Source code in src/mfdro/signal.py

@classmethod
def load(cls, directory: str | Path) -> SignalPath:
    """Load and integrity-check a path written by :meth:`save`.

    Format-1 paths remain readable. Their per-frequency distances cannot
    be reconstructed from the aggregate signal, so the migrated columns
    contain ``NaN``. Saving the returned object writes the current format.
    """

    source = Path(directory).expanduser().resolve()
    manifest_path = source.joinpath("manifest.json")
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError) as exc:
        raise DataContractError("Signal path manifest is missing or invalid.") from exc
    if not isinstance(manifest, Mapping):
        raise DataContractError("Signal path manifest must contain one object.")
    raw_format_version = manifest.get("format_version")
    if (
        isinstance(raw_format_version, bool)
        or not isinstance(raw_format_version, int)
        or raw_format_version not in SUPPORTED_PATH_FORMAT_VERSIONS
    ):
        raise DataContractError(
            f"Unsupported signal path format version: {raw_format_version!r}."
        )
    raw_hashes = manifest.get("sha256")
    if not isinstance(raw_hashes, Mapping):
        raise DataContractError("Signal path manifest does not contain file checksums.")

    payloads: dict[str, str] = {}
    for name in ("config.json", "estimates.json", "audit.json", "skipped.json"):
        try:
            payload = source.joinpath(name).read_text(encoding="utf-8")
        except FileNotFoundError as exc:
            raise DataContractError(f"Signal path file is missing: {name}.") from exc
        expected_hash = raw_hashes.get(name)
        observed_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
        if expected_hash != observed_hash:
            raise DataContractError(f"Signal path checksum differs for {name}.")
        payloads[name] = payload

    config = SignalConfig.from_json(payloads["config.json"])
    if manifest.get("config_digest") != config.digest:
        raise DataContractError("Signal path configuration digest does not match its manifest.")
    estimates = _frame_from_json(payloads["estimates.json"])
    audit = _frame_from_json(payloads["audit.json"])
    skipped = _frame_from_json(payloads["skipped.json"])
    _validate_manifest_rows(
        manifest,
        {"estimates": estimates, "audit": audit, "skipped": skipped},
    )
    if raw_format_version == 1:
        estimates = _upgrade_v1_estimates(estimates, config.frequencies)
    result = cls(
        estimates=estimates,
        audit=audit,
        config=config,
        skipped=skipped,
    )
    _validate_loaded_path(result)
    return result