Strategies¶

fundcloud.strategies exposes the abstract BaseStrategy together with two preset implementations (Hold, DCA), a per-bar Context object, and the Cadence / Scheduler primitives that govern when a strategy fires. The @register_strategy decorator participates in Catalog serialisation so a strategy name round-trips through YAML configs. For usage patterns and the custom-strategy worked example, see the DCA & Hold guide.

fundcloud.strategies ¶

Decision-logic presets and extension points.

Two built-in presets (:class:Hold and :class:DCA) cover the common retail / allocation workflows. Custom strategies subclass :class:BaseStrategy and may optionally register themselves via :func:register_strategy so they become discoverable by name.

BaseStrategy ¶

Bases: ABC

ABC for every concrete strategy.

close ¶

close(portfolio: Portfolio) -> None

End-of-run hook.

Default: no-op. Not abstract because most strategies don't need it.

Source code in python/fundcloud/strategies/base.py

def close(self, portfolio: Portfolio) -> None:  # noqa: B027
    """End-of-run hook.

    Default: no-op. Not abstract because most strategies don't need it.
    """

decide `abstractmethod` ¶

decide(ctx: Context) -> list[Order]

Return the orders this strategy wants executed next bar.

Source code in python/fundcloud/strategies/base.py

@abstractmethod
def decide(self, ctx: Context) -> list[Order]:
    """Return the orders this strategy wants executed **next bar**."""

init ¶

init(bars: DataFrame, portfolio: Portfolio) -> None

One-shot setup called before the first decide.

Default: no-op. Not abstract because most strategies don't need warm-up.

Source code in python/fundcloud/strategies/base.py

def init(self, bars: pd.DataFrame, portfolio: Portfolio) -> None:  # noqa: B027
    """One-shot setup called before the first ``decide``.

    Default: no-op. Not abstract because most strategies don't need warm-up.
    """

Context `dataclass` ¶

Context(
    ts: Timestamp,
    bar: Series,
    history: DataFrame,
    portfolio: Portfolio,
    assets: tuple[str, ...],
    extras: Mapping[str, Any] = dict(),
)

Per-bar context handed to :meth:BaseStrategy.decide.

Attributes:

Name	Type	Description
`ts`	`Timestamp`	Current bar timestamp.
`bar`	`Series`	Current bar. For a `Bars` MultiIndex frame this is a `Series` indexed by `(field, asset)`; for a simple price panel it's a single-level Series indexed by asset.
`history`	`DataFrame`	Bars up to and including `ts`. Strategies look back into this for indicators.
`portfolio`	`Portfolio`	Live :class:`Portfolio` — callers should treat this as read-only.
`assets`	`tuple[str, ...]`	Convenience: asset universe ordered by column appearance.
`extras`	`Mapping[str, Any]`	Optional dict for user-specified scheduled events, factor scores, etc. Populated by the simulator if needed, else an empty dict.

Hold ¶

Hold(
    weights: WeightsLike,
    *,
    rebalance: RebalanceSpec | None = None,
    start: Timestamp | str | None = None,
)

Bases: BaseStrategy

Allocate to target weights once, hold; optionally rebalance.

Parameters:

Name	Type	Description	Default
`weights`	`WeightsLike`	Either a mapping of `asset -> weight` or a callable receiving the `init` warm-up window and returning such a mapping. Weights must sum to 1.	required
`rebalance`	`RebalanceSpec \| None`	If supplied, restore target weights at each cadence boundary.	`None`
`start`	`Timestamp \| str \| None`	Optional lock-out: don't place the first allocation before `start`.	`None`

Examples:

Buy-and-hold 60/40 equity / bonds, no rebalancing (weights drift with prices):

>>> from fundcloud.strategies import Hold
>>> Hold({"SPY": 0.6, "AGG": 0.4})
<fundcloud.strategies.hold.Hold object at ...>

Quarterly-rebalanced 60/40 with a 5 %-drift tolerance:

>>> from fundcloud.strategies import RebalanceSpec
>>> Hold(
...     {"SPY": 0.6, "AGG": 0.4},
...     rebalance=RebalanceSpec(horizon="91D", tolerance=0.05),
... )
<fundcloud.strategies.hold.Hold object at ...>

Source code in python/fundcloud/strategies/hold.py

def __init__(
    self,
    weights: WeightsLike,
    *,
    rebalance: RebalanceSpec | None = None,
    start: pd.Timestamp | str | None = None,
) -> None:
    self._weights_spec = weights
    self._rebalance = rebalance
    self._start = pd.Timestamp(start) if start is not None else None
    self._resolved_weights: dict[str, float] = {}
    self._triggered_once: bool = False
    self._rebalance_triggers: set[pd.Timestamp] = set()

RebalanceSpec `dataclass` ¶

RebalanceSpec(
    horizon: str = "monthly", tolerance: float = 0.0
)

Optional rebalance policy for :class:Hold.

DCA ¶

DCA(
    amount: float | Mapping[str, float],
    *,
    horizon: HorizonName | Cadence | str = "monthly",
    weights: Mapping[str, float] | None = None,
    start: Timestamp | str | None = None,
    end: Timestamp | str | None = None,
    sell_on_end: bool = False,
)

Bases: BaseStrategy

Invest a fixed amount at a fixed cadence.

Parameters:

Name	Type	Description	Default
`amount`	`float \| Mapping[str, float]`	Either a scalar (distributed across `weights`) or a mapping `asset -> dollars`.	required
`horizon`	`HorizonName \| Cadence \| str`	Cadence — `"daily"`, `"weekly"` (7 calendar days), `"monthly"`, or a :class:`Cadence` for arbitrary steps.	`'monthly'`
`weights`	`Mapping[str, float] \| None`	Optional. When omitted with a scalar `amount`, DCA spreads the deposit equally across every asset in the `bars` frame it sees at :meth:`init`. Provide an explicit mapping (fractions summing to 1) to weight the split unevenly.	`None`
`start`	`Timestamp \| str \| None`	Optional window inside which DCA fires.	`None`
`end`	`Timestamp \| str \| None`	Optional window inside which DCA fires.	`None`
`sell_on_end`	`bool`	When `True`, close all positions on the last fire after `end`.	`False`

Examples:

Single-asset weekly DCA into SPY — the classic retail deposit:

>>> from fundcloud.strategies import DCA
>>> DCA(amount=500.0, horizon="weekly", weights={"SPY": 1.0})
<fundcloud.strategies.dca.DCA object at ...>

Multi-asset monthly allocation with explicit dollar buckets per leg:

>>> DCA({"SPY": 300.0, "AGG": 200.0}, horizon="monthly")
<fundcloud.strategies.dca.DCA object at ...>

Scalar amount with no weights — equal-weight over whatever assets the bars frame contains:

>>> DCA(500.0, horizon="weekly")
<fundcloud.strategies.dca.DCA object at ...>

Source code in python/fundcloud/strategies/dca.py

def __init__(
    self,
    amount: float | Mapping[str, float],
    *,
    horizon: HorizonName | Cadence | str = "monthly",
    weights: Mapping[str, float] | None = None,
    start: pd.Timestamp | str | None = None,
    end: pd.Timestamp | str | None = None,
    sell_on_end: bool = False,
) -> None:
    self._scalar_amount: float | None
    if isinstance(amount, Mapping):
        self._amounts: dict[str, float] = {k: float(v) for k, v in amount.items()}
        self._scalar_amount = None
    else:
        self._scalar_amount = float(amount)
        if weights is None:
            # Defer the per-asset split to init(), where we can read
            # the bars frame and divide equally across its assets.
            self._amounts = {}
        else:
            total_w = sum(weights.values())
            if abs(total_w - 1.0) > 1e-6:
                msg = f"DCA weights must sum to 1, got {total_w}"
                raise ValueError(msg)
            self._amounts = {k: self._scalar_amount * float(v) for k, v in weights.items()}
    self._horizon = horizon
    self._start = pd.Timestamp(start) if start is not None else None
    self._end = pd.Timestamp(end) if end is not None else None
    self._sell_on_end = sell_on_end
    self._fire_set: set[pd.Timestamp] = set()
    self._last_fire: pd.Timestamp | None = None
    self._ended: bool = False

Scheduler ¶

Factory for :class:Cadence presets.

from_horizon `staticmethod` ¶

from_horizon(
    horizon: HorizonName | Cadence | str,
    *,
    anchor: Timestamp | None = None,
) -> Cadence

Resolve a user-facing horizon into a concrete :class:Cadence.

Accepted forms:

"daily" → every trading day (1D step anchored at anchor).
"weekly" → every 7 calendar days from anchor (not ISO weekday 1 — matches PRD's "(7 days)" wording).
"monthly" → same day-of-month as anchor each month, falling back to the last trading day of the month if missing.
a :class:Cadence is returned as-is (with anchor merged in).
any pandas offset string (e.g. "30D", "3W") → :class:Cadence.

Source code in python/fundcloud/strategies/scheduler.py

@staticmethod
def from_horizon(
    horizon: HorizonName | Cadence | str,
    *,
    anchor: pd.Timestamp | None = None,
) -> Cadence:
    """Resolve a user-facing horizon into a concrete :class:`Cadence`.

    Accepted forms:

    * ``"daily"``  → every trading day (``1D`` step anchored at ``anchor``).
    * ``"weekly"`` → every 7 calendar days from ``anchor``
      (**not** ISO weekday 1 — matches PRD's "(7 days)" wording).
    * ``"monthly"`` → same day-of-month as ``anchor`` each month,
      falling back to the last trading day of the month if missing.
    * a :class:`Cadence` is returned as-is (with ``anchor`` merged in).
    * any pandas offset string (e.g. ``"30D"``, ``"3W"``) → :class:`Cadence`.
    """
    if isinstance(horizon, Cadence):
        return Cadence(step=horizon.step, anchor=anchor or horizon.anchor)
    if horizon == "daily":
        return Cadence(step="1D", anchor=anchor)
    if horizon == "weekly":
        return Cadence(step="7D", anchor=anchor)
    if horizon == "monthly":
        return _MonthlyCadence(anchor=anchor)  # type: ignore[return-value]
    # Treat anything else as a pandas-parseable offset.
    try:
        pd.Timedelta(horizon)
    except ValueError as e:  # pragma: no cover
        msg = f"unknown horizon: {horizon!r}"
        raise ValueError(msg) from e
    return Cadence(step=str(horizon), anchor=anchor)

Cadence `dataclass` ¶

Cadence(step: str = '1D', anchor: Timestamp | None = None)

An explicit cadence step. Use :meth:Scheduler.from_horizon for presets.

step is any pandas-parseable offset like "7D", "14D", or "1D". Anchor is the first timestamp the cadence fires on; if unspecified the first trigger falls on the first timestamp in the passed index.

triggers ¶

triggers(
    index: DatetimeIndex,
    *,
    start: Timestamp | str | None = None,
    end: Timestamp | str | None = None,
) -> pd.DatetimeIndex

Return the subset of index that the cadence fires on.

Source code in python/fundcloud/strategies/scheduler.py

def triggers(
    self,
    index: pd.DatetimeIndex,
    *,
    start: pd.Timestamp | str | None = None,
    end: pd.Timestamp | str | None = None,
) -> pd.DatetimeIndex:
    """Return the subset of ``index`` that the cadence fires on."""
    if len(index) == 0:
        return pd.DatetimeIndex([])
    effective_start = pd.Timestamp(start) if start is not None else index[0]
    effective_end = pd.Timestamp(end) if end is not None else index[-1]
    anchor = self.anchor or effective_start

    step = pd.Timedelta(self.step)
    fires: list[pd.Timestamp] = []
    cursor = anchor
    while cursor <= effective_end:
        if cursor >= effective_start:
            # Snap to the next available bar in ``index`` (inclusive).
            pos = index.searchsorted(cursor, side="left")
            if pos >= len(index):
                break
            snapped = index[pos]
            if snapped <= effective_end and (not fires or snapped > fires[-1]):
                fires.append(snapped)
        cursor = cursor + step
    return pd.DatetimeIndex(fires)

register_strategy ¶

register_strategy(name: str) -> Any

Decorator: make a strategy class discoverable by name.

Source code in python/fundcloud/strategies/base.py

def register_strategy(name: str) -> Any:
    """Decorator: make a strategy class discoverable by name."""

    def deco(cls: type[BaseStrategy]) -> type[BaseStrategy]:
        _REGISTRY[name] = cls
        return cls

    return deco

registered_strategies ¶

registered_strategies() -> dict[str, type[BaseStrategy]]

Source code in python/fundcloud/strategies/base.py

def registered_strategies() -> dict[str, type[BaseStrategy]]:
    return dict(_REGISTRY)

Strategies¶

fundcloud.strategies ¶

BaseStrategy ¶

close ¶

decide abstractmethod ¶

init ¶

Context dataclass ¶

Hold ¶

RebalanceSpec dataclass ¶

DCA ¶

Scheduler ¶

from_horizon staticmethod ¶

Cadence dataclass ¶

triggers ¶

register_strategy ¶

registered_strategies ¶

decide `abstractmethod` ¶

Context `dataclass` ¶

RebalanceSpec `dataclass` ¶

from_horizon `staticmethod` ¶

Cadence `dataclass` ¶