Configuration API

FrequencySpec

mfdro.frequency.FrequencySpec(name: str, horizon: float, rule: str | None = None, closed: Side | None = None, label: Side | None = None, origin: str = 'start_day', offset: str | None = None, min_observations: int = 1) dataclass

Define one empirical return frequency.

Parameters:

Name	Type	Description	Default
name	str	Stable public name used in input mappings and output columns.	required
horizon	float	Effective number of base periods used by horizon scaling.	required
rule	str | None	pandas offset alias used to aggregate the base panel. The first frequency must use None because it represents the unaggregated input.	None
closed	Side | None	Explicit pandas resampling conventions. Their values are included in the scientific configuration digest.	None
label	Side | None	Explicit pandas resampling conventions. Their values are included in the scientific configuration digest.	None
origin	Side | None	Explicit pandas resampling conventions. Their values are included in the scientific configuration digest.	None
offset	Side | None	Explicit pandas resampling conventions. Their values are included in the scientific configuration digest.	None
min_observations	int	Minimum number of base observations required in an aggregation bin.	1

to_dict() -> dict[str, object]

Return a JSON-serialisable frequency definition.

Source code in src/mfdro/frequency.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-serialisable frequency definition."""

    return {
        "name": self.name,
        "horizon": self.horizon,
        "rule": self.rule,
        "closed": self.closed,
        "label": self.label,
        "origin": self.origin,
        "offset": self.offset,
        "min_observations": self.min_observations,
    }

from_dict(payload: Mapping[str, object]) -> FrequencySpec classmethod

Construct a frequency definition from :meth:to_dict output.

Source code in src/mfdro/frequency.py

@classmethod
def from_dict(cls, payload: Mapping[str, object]) -> FrequencySpec:
    """Construct a frequency definition from :meth:`to_dict` output."""

    if not isinstance(payload, Mapping):
        raise ConfigurationError("A frequency payload must be a mapping.")
    expected = {
        "name",
        "horizon",
        "rule",
        "closed",
        "label",
        "origin",
        "offset",
        "min_observations",
    }
    observed = set(payload)
    if observed != expected:
        missing = sorted(expected - observed)
        extra = sorted(observed - expected)
        raise ConfigurationError(f"Frequency fields differ: missing={missing}, extra={extra}.")
    return cls(
        name=cast(str, payload["name"]),
        horizon=cast(float, payload["horizon"]),
        rule=cast(str | None, payload["rule"]),
        closed=cast(Side | None, payload["closed"]),
        label=cast(Side | None, payload["label"]),
        origin=cast(str, payload["origin"]),
        offset=cast(str | None, payload["offset"]),
        min_observations=cast(int, payload["min_observations"]),
    )

SignalConfig

mfdro.config.SignalConfig(frequency_specs: tuple[FrequencySpec, ...] = DEFAULT_FREQUENCY_SPECS, scaling: Scaling = 'power', scaling_exponent: float = 0.5, frequency_weighting: FrequencyWeighting = 'uniform', explicit_frequency_weights: tuple[float, ...] | None = None, barycenter: Barycenter = 'free_support', barycenter_size: int = 50, barycenter_weights: tuple[float, ...] | None = None, barycenter_random_state: int = 0, distance: Distance = 'sliced', n_projections: int = 200, n_quantiles: int = 200, random_state: int = 20250301, barycenter_max_iter: int = 30, barycenter_tolerance: float = 0.0001) dataclass

Configuration of a multi-frequency ambiguity-signal estimate.

frequency_specs is the only frequency interface. Specifications with rule=None can represent measures constructed by the caller; the first specification must always be the unaggregated base panel. Walk-forward estimation additionally requires a resampling rule on every later specification.

Parameters:

Name	Type	Description	Default
frequency_specs	tuple[FrequencySpec, ...]	Ordered empirical-frequency definitions. At least two are required.	DEFAULT_FREQUENCY_SPECS
scaling	Scaling	"power" for horizon scaling or "realized_volatility" for within-frequency asset standardization.	'power'
scaling_exponent	float	Exponent applied to horizons under power scaling.	0.5
frequency_weighting	FrequencyWeighting	Rule used to aggregate dispersion across frequencies.	'uniform'
explicit_frequency_weights	tuple[float, ...] | None	One positive dispersion weight per frequency when weighting is explicit.	None
barycenter	Barycenter	Multivariate free-support or projected-quantile construction.	'free_support'
barycenter_size	int	Number of free-support atoms. Ignored by "projected_quantile".	50
barycenter_weights	tuple[float, ...] | None	Optional positive measure weights used to construct the center.	None
barycenter_random_state	int	uint32 seed for weighted k-means++ support initialization.	0
distance	Distance	Sliced approximation or exact discrete squared transport cost.	'sliced'
n_projections	int	Number of random directions used by projected calculations.	200
n_quantiles	int	Quantile-grid size used by projected calculations.	200
random_state	int	Base uint32 seed used directly or to derive walk-forward seeds.	20250301
barycenter_max_iter	int	Maximum free-support solver iterations.	30
barycenter_tolerance	float	Positive stopping threshold for the free-support solver.	0.0001

Notes

Sequence inputs are copied into tuples. digest therefore remains stable if caller-owned lists are mutated after construction.

frequencies: tuple[str, ...] property

Return canonical frequency names in numerical coordinate order.

horizons: tuple[float, ...] property

Return effective base-period horizons in frequency order.

frequency_grid: tuple[FrequencySpec, ...] property

Return the canonical, fully materialised frequency specification.

digest: str property

Return a stable SHA-256 identity for this scientific configuration.

reference() -> SignalConfig classmethod

Return the fully explicit reference research configuration.

Source code in src/mfdro/config.py

@classmethod
def reference(cls) -> SignalConfig:
    """Return the fully explicit reference research configuration."""

    return cls()

projected(*, frequency_specs: Sequence[FrequencySpec] = DEFAULT_FREQUENCY_SPECS, n_projections: int = 100, n_quantiles: int = 100, random_state: int = 20250301) -> SignalConfig classmethod

Return a lightweight projected-quantile configuration for exploration.

Source code in src/mfdro/config.py

@classmethod
def projected(
    cls,
    *,
    frequency_specs: Sequence[FrequencySpec] = DEFAULT_FREQUENCY_SPECS,
    n_projections: int = 100,
    n_quantiles: int = 100,
    random_state: int = 20250301,
) -> SignalConfig:
    """Return a lightweight projected-quantile configuration for exploration."""

    return cls(
        frequency_specs=tuple(frequency_specs),
        barycenter="projected_quantile",
        n_projections=n_projections,
        n_quantiles=n_quantiles,
        random_state=random_state,
    )

with_updates(**changes: object) -> SignalConfig

Return a validated copy with selected fields changed.

This is the ergonomic route for modifying an immutable preset without repeating every unchanged field. Unknown field names fail explicitly.

Source code in src/mfdro/config.py

def with_updates(self, **changes: object) -> SignalConfig:
    """Return a validated copy with selected fields changed.

    This is the ergonomic route for modifying an immutable preset without
    repeating every unchanged field. Unknown field names fail explicitly.
    """

    try:
        return replace(self, **changes)  # type: ignore[arg-type]
    except TypeError as exc:
        unknown = sorted(set(changes) - set(self.__dataclass_fields__))
        if unknown:
            raise ConfigurationError(f"Unknown configuration fields: {unknown}.") from exc
        raise

to_dict() -> dict[str, object]

Return a versioned, JSON-serialisable scientific configuration.

Source code in src/mfdro/config.py

def to_dict(self) -> dict[str, object]:
    """Return a versioned, JSON-serialisable scientific configuration."""

    return {
        "schema_version": CONFIG_SCHEMA_VERSION,
        "frequency_specs": [spec.to_dict() for spec in self.frequency_specs],
        "scaling": self.scaling,
        "scaling_exponent": self.scaling_exponent,
        "frequency_weighting": self.frequency_weighting,
        "explicit_frequency_weights": (
            None
            if self.explicit_frequency_weights is None
            else list(self.explicit_frequency_weights)
        ),
        "barycenter": self.barycenter,
        "barycenter_size": self.barycenter_size,
        "barycenter_weights": (
            None if self.barycenter_weights is None else list(self.barycenter_weights)
        ),
        "barycenter_random_state": self.barycenter_random_state,
        "distance": self.distance,
        "n_projections": self.n_projections,
        "n_quantiles": self.n_quantiles,
        "random_state": self.random_state,
        "barycenter_max_iter": self.barycenter_max_iter,
        "barycenter_tolerance": self.barycenter_tolerance,
    }

from_dict(payload: Mapping[str, object]) -> SignalConfig classmethod

Construct a configuration from :meth:to_dict output.

Source code in src/mfdro/config.py

@classmethod
def from_dict(cls, payload: Mapping[str, object]) -> SignalConfig:
    """Construct a configuration from :meth:`to_dict` output."""

    if not isinstance(payload, Mapping):
        raise ConfigurationError("Configuration payload must be a mapping.")
    expected = {
        "schema_version",
        "frequency_specs",
        "scaling",
        "scaling_exponent",
        "frequency_weighting",
        "explicit_frequency_weights",
        "barycenter",
        "barycenter_size",
        "barycenter_weights",
        "barycenter_random_state",
        "distance",
        "n_projections",
        "n_quantiles",
        "random_state",
        "barycenter_max_iter",
        "barycenter_tolerance",
    }
    observed = set(payload)
    if observed != expected:
        missing = sorted(expected - observed)
        extra = sorted(observed - expected)
        raise ConfigurationError(
            f"Configuration fields differ: missing={missing}, extra={extra}."
        )
    if payload["schema_version"] != CONFIG_SCHEMA_VERSION:
        raise ConfigurationError(
            f"Unsupported configuration schema version: {payload['schema_version']!r}."
        )
    raw_specs = cls._as_tuple(payload["frequency_specs"], "frequency_specs")
    specs = tuple(
        FrequencySpec.from_dict(cast(Mapping[str, object], item)) for item in raw_specs
    )
    return cls(
        frequency_specs=specs,
        scaling=cast(Scaling, payload["scaling"]),
        scaling_exponent=cast(float, payload["scaling_exponent"]),
        frequency_weighting=cast(FrequencyWeighting, payload["frequency_weighting"]),
        explicit_frequency_weights=cast(
            tuple[float, ...] | None,
            payload["explicit_frequency_weights"],
        ),
        barycenter=cast(Barycenter, payload["barycenter"]),
        barycenter_size=cast(int, payload["barycenter_size"]),
        barycenter_weights=cast(tuple[float, ...] | None, payload["barycenter_weights"]),
        barycenter_random_state=cast(int, payload["barycenter_random_state"]),
        distance=cast(Distance, payload["distance"]),
        n_projections=cast(int, payload["n_projections"]),
        n_quantiles=cast(int, payload["n_quantiles"]),
        random_state=cast(int, payload["random_state"]),
        barycenter_max_iter=cast(int, payload["barycenter_max_iter"]),
        barycenter_tolerance=cast(float, payload["barycenter_tolerance"]),
    )

to_json(*, indent: int | None = 2) -> str

Serialise the configuration to deterministic JSON text.

Source code in src/mfdro/config.py

def to_json(self, *, indent: int | None = 2) -> str:
    """Serialise the configuration to deterministic JSON text."""

    return json.dumps(self.to_dict(), indent=indent, sort_keys=True) + "\n"

from_json(payload: str) -> SignalConfig classmethod

Construct a configuration from JSON text.

Source code in src/mfdro/config.py

@classmethod
def from_json(cls, payload: str) -> SignalConfig:
    """Construct a configuration from JSON text."""

    if not isinstance(payload, str):
        raise ConfigurationError("Configuration JSON must be text.")
    try:
        decoded = json.loads(payload)
    except json.JSONDecodeError as exc:
        raise ConfigurationError("Configuration JSON is invalid.") from exc
    if not isinstance(decoded, Mapping):
        raise ConfigurationError("Configuration JSON must contain one object.")
    return cls.from_dict(decoded)

write_json(path: str | Path) -> Path

Write deterministic configuration JSON and return the resolved path.

Source code in src/mfdro/config.py

def write_json(self, path: str | Path) -> Path:
    """Write deterministic configuration JSON and return the resolved path."""

    destination = Path(path).expanduser().resolve()
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(self.to_json(), encoding="utf-8")
    return destination

read_json(path: str | Path) -> SignalConfig classmethod

Read a configuration written by :meth:write_json.

Source code in src/mfdro/config.py

@classmethod
def read_json(cls, path: str | Path) -> SignalConfig:
    """Read a configuration written by :meth:`write_json`."""

    source = Path(path).expanduser().resolve()
    return cls.from_json(source.read_text(encoding="utf-8"))