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.

The simulator drives subclasses through a three-stage lifecycle: :meth:init once before the first bar, :meth:decide once per bar, :meth:close once after the last bar. Only :meth:decide is required.

Examples:

Minimal momentum strategy that buys 1 share when the 10-day return is positive and sells when it's negative:

>>> import pandas as pd
>>> from fundcloud.strategies import BaseStrategy, Context
>>> from fundcloud.sim import Order
>>>
>>> class Momentum(BaseStrategy):
...     def __init__(self, lookback: int = 10) -> None:
...         self.lookback = lookback
...
...     def decide(self, ctx: Context) -> list[Order]:
...         if len(ctx.history) < self.lookback:
...             return []
...         orders: list[Order] = []
...         close = ctx.history.xs("close", axis=1, level=0)
...         ret = close.iloc[-1] / close.iloc[-self.lookback] - 1.0
...         for asset, r in ret.items():
...             side = "buy" if r > 0 else "sell"
...             orders.append(Order(ts=ctx.ts, asset=str(asset), side=side, qty=1.0))
...         return orders

name `property` ¶

name: str

Class name — used as the default label in :class:SimResult.

close ¶

close(portfolio: Portfolio) -> None

End-of-run hook.

Default: no-op. Override to compute summary stats, persist diagnostics, or release external resources. Not abstract because most strategies don't need it.

Parameters:

Name	Type	Description	Default
`portfolio`	`Portfolio`	Final :class:`~fundcloud.portfolio.Portfolio` state after the last bar.	required

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

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

    Default: no-op. Override to compute summary stats, persist
    diagnostics, or release external resources. Not abstract
    because most strategies don't need it.

    Parameters
    ----------
    portfolio
        Final :class:`~fundcloud.portfolio.Portfolio` state after
        the last bar.
    """

decide `abstractmethod` ¶

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

Return the orders this strategy wants executed.

Called once per bar. The simulator queues the returned orders for execution under its :class:~fundcloud.sim.ExecutionModel (default :class:~fundcloud.sim.NextBarOpen, so orders emitted on bar t fill at the open of bar t + 1).

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	Per-bar :class:`Context` — current timestamp, bar slice, full history through `ctx.ts`, live portfolio, asset universe.	required

Returns:

Type	Description
`list of Order`	Zero or more :class:`~fundcloud.sim.Order` instances. An empty list means "do nothing this bar".

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

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

    Called once per bar. The simulator queues the returned orders
    for execution under its :class:`~fundcloud.sim.ExecutionModel`
    (default :class:`~fundcloud.sim.NextBarOpen`, so orders emitted
    on bar *t* fill at the open of bar *t + 1*).

    Parameters
    ----------
    ctx
        Per-bar :class:`Context` — current timestamp, bar slice,
        full history through ``ctx.ts``, live portfolio, asset
        universe.

    Returns
    -------
    list of Order
        Zero or more :class:`~fundcloud.sim.Order` instances. An
        empty list means "do nothing this bar".
    """

init ¶

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

One-shot setup called before the first :meth:decide.

Default: no-op. Override to compute a warm-up window, schedule cadences, or stash any state derived from the full bars frame. Not abstract because most strategies don't need it.

Parameters:

Name	Type	Description	Default
`bars`	`DataFrame`	The full Bars frame the simulator will iterate over — `(field, symbol)` MultiIndex columns, sorted DatetimeIndex. Use it to precompute schedules, asset lists, etc.	required
`portfolio`	`Portfolio`	The live :class:`~fundcloud.portfolio.Portfolio` (already funded with starting cash). Treat as read-only here.	required

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 :meth:`decide`.

    Default: no-op. Override to compute a warm-up window, schedule
    cadences, or stash any state derived from the full ``bars``
    frame. Not abstract because most strategies don't need it.

    Parameters
    ----------
    bars
        The full Bars frame the simulator will iterate over —
        ``(field, symbol)`` MultiIndex columns, sorted DatetimeIndex.
        Use it to precompute schedules, asset lists, etc.
    portfolio
        The live :class:`~fundcloud.portfolio.Portfolio` (already
        funded with starting cash). Treat as read-only here.
    """

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.

Strategies are stateless on the bar boundary (they may carry their own state across bars, but the simulator hands them everything they need to decide on this bar via this object).

Attributes:

Name	Type	Description
`ts`	`Timestamp`	Current bar timestamp.
`bar`	`Series`	Current bar. For a Bars MultiIndex frame this is a :class:`pandas.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 (rolling means, momentum windows, …).
`portfolio`	`Portfolio`	Live :class:`~fundcloud.portfolio.Portfolio`. Treat as read-only — the simulator owns all state mutations.
`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 when configured, else an empty dict.

Hold ¶

Hold(
    weights: WeightsLike | None = None,
    *,
    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 \| None`	Target allocation. May be a `Mapping[str, float]`, a :class:`pandas.Series`, or a callable receiving the `init` warm-up bars frame and returning such a mapping. Weights must sum to 1. When omitted (`None`), Hold spreads the position equally across every asset in the bars frame it sees at :meth:`init`. Useful for a quick equal-weight baseline without having to enumerate the universe up front.	`None`
`rebalance`	`RebalanceSpec \| None`	If supplied, restore target weights at each cadence boundary. See :class:`RebalanceSpec` for the cadence + tolerance knobs.	`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 ...>

Equal-weight default — distribute evenly over whatever assets are in the bars frame:

>>> Hold()
<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 | None = None,
    *,
    rebalance: RebalanceSpec | None = None,
    start: pd.Timestamp | str | None = None,
) -> None:
    self._weights_spec: WeightsLike | None = 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] | None = None,
    *,
    amount_pct: float | Mapping[str, float] | None = None,
    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.

Exactly one of amount or amount_pct must be provided — they're two ways to spell the same thing:

amount — explicit dollars per fire (scalar or per-asset).
amount_pct — fraction of the current portfolio equity deployed at each fire (scalar or per-asset). The dollar size is recomputed at every fire from Portfolio.equity_curve, so the deposit grows or shrinks with the portfolio. Each fire is also clipped to currently-available cash — DCA never borrows; once cash is exhausted, subsequent fires emit no orders. On the very first fire (no equity history yet) it falls back to starting cash.

Parameters:

Name	Type	Description	Default
`amount`	`float \| Mapping[str, float] \| None`	Dollars per fire. Either a scalar (distributed across `weights`) or a mapping `asset -> dollars`. Mutually exclusive with `amount_pct`.	`None`
`amount_pct`	`float \| Mapping[str, float] \| None`	Equity fraction per fire. Either a scalar in `[0, 1]` (distributed across `weights`) or a mapping `asset -> fraction`. Mutually exclusive with `amount`.	`None`
`horizon`	`HorizonName \| Cadence \| str`	Cadence — `"daily"`, `"weekly"` (7 calendar days), `"monthly"`, `"quarterly"` (every 3 months, same-day-of-month snap rule as monthly), or a :class:`Cadence` for arbitrary steps.	`'monthly'`
`weights`	`Mapping[str, float] \| None`	Optional. When omitted with a scalar `amount` / `amount_pct`, DCA spreads the deposit equally across every asset in the `bars` frame at :meth:`init`. Provide an explicit mapping (fractions summing to 1) to weight the split unevenly. Negative weights are allowed and produce short-sells: e.g. `weights={"A": 1.5, "B": -0.5}` says long A by 1.5x the deposit and short B by 0.5x — the short proceeds fund the oversized long, and net deployed cash equals the deposit.	`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 bar after 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 ...>

Percentage of current equity instead of fixed dollars — deploy 1 % of the portfolio each month, scaling automatically as equity grows:

>>> DCA(amount_pct=0.01, horizon="monthly")
<fundcloud.strategies.dca.DCA object at ...>

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

def __init__(
    self,
    amount: float | Mapping[str, float] | None = None,
    *,
    amount_pct: float | Mapping[str, float] | None = None,
    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:
    if (amount is None) == (amount_pct is None):
        msg = "DCA: provide exactly one of `amount` or `amount_pct`."
        raise ValueError(msg)

    # Resolve the active sizing source into matching scalar / per-asset
    # storage. Per-asset mappings are stored verbatim; a scalar with no
    # weights defers the equal split to init() (where we can read the
    # bars frame); a scalar with weights is pre-multiplied here.
    self._scalar_amount: float | None = None
    self._amounts: dict[str, float] = {}
    self._scalar_amount_pct: float | None = None
    self._amount_pcts: dict[str, float] = {}

    if amount is not None:
        if isinstance(amount, Mapping):
            if weights is not None:
                msg = "DCA: `weights` must be omitted when `amount` is a mapping."
                raise ValueError(msg)
            self._amounts = {k: float(v) for k, v in amount.items()}
        else:
            self._scalar_amount = float(amount)
            if weights is None:
                self._amounts = {}
            else:
                _validate_weights_sum_to_one(weights)
                self._amounts = {k: self._scalar_amount * float(v) for k, v in weights.items()}
    else:
        assert amount_pct is not None  # for type-checkers; guarded above
        _validate_amount_pct_values(amount_pct)
        if isinstance(amount_pct, Mapping):
            if weights is not None:
                msg = "DCA: `weights` must be omitted when `amount_pct` is a mapping."
                raise ValueError(msg)
            self._amount_pcts = {k: float(v) for k, v in amount_pct.items()}
        else:
            self._scalar_amount_pct = float(amount_pct)
            if weights is None:
                self._amount_pcts = {}
            else:
                _validate_weights_sum_to_one(weights)
                self._amount_pcts = {
                    k: self._scalar_amount_pct * 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.
"quarterly" → same rule as monthly, stepping every 3 months.
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.
    * ``"quarterly"`` → same rule as monthly, stepping every 3 months.
    * 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]
    if horizon == "quarterly":
        return _QuarterlyCadence(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.

Registered strategies can be looked up via :func:registered_strategies — useful for config-driven runners that pick the strategy class from a string.

Parameters:

Name	Type	Description	Default
`name`	`str`	String key the class will be stored under. Typically a short snake-case identifier.	required

Returns:

Type	Description
`callable`	The decorator that registers and returns the class unchanged.

Examples:

>>> from fundcloud.strategies import BaseStrategy, register_strategy, registered_strategies
>>> @register_strategy("noop")
... class Noop(BaseStrategy):
...     def decide(self, ctx):
...         return []
>>> "noop" in registered_strategies()
True

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

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

    Registered strategies can be looked up via
    :func:`registered_strategies` — useful for config-driven runners
    that pick the strategy class from a string.

    Parameters
    ----------
    name
        String key the class will be stored under. Typically a short
        snake-case identifier.

    Returns
    -------
    callable
        The decorator that registers and returns the class unchanged.

    Examples
    --------
    >>> from fundcloud.strategies import BaseStrategy, register_strategy, registered_strategies
    >>> @register_strategy("noop")
    ... class Noop(BaseStrategy):
    ...     def decide(self, ctx):
    ...         return []
    >>> "noop" in registered_strategies()
    True
    """

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

    return deco

registered_strategies ¶

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

Return a snapshot of the strategy registry.

Returns:

Type	Description
`dict[str, type[BaseStrategy]]`	Copy of the `name -> class` mapping populated by :func:`register_strategy`. Mutating the returned dict does not affect the registry.

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

def registered_strategies() -> dict[str, type[BaseStrategy]]:
    """Return a snapshot of the strategy registry.

    Returns
    -------
    dict[str, type[BaseStrategy]]
        Copy of the ``name -> class`` mapping populated by
        :func:`register_strategy`. Mutating the returned dict does not
        affect the registry.
    """
    return dict(_REGISTRY)

Strategies¶

fundcloud.strategies ¶

BaseStrategy ¶

name property ¶

close ¶

decide abstractmethod ¶

init ¶

Context dataclass ¶

Hold ¶

RebalanceSpec dataclass ¶

DCA ¶

Scheduler ¶

from_horizon staticmethod ¶

Cadence dataclass ¶

triggers ¶

register_strategy ¶

registered_strategies ¶

name `property` ¶

decide `abstractmethod` ¶

Context `dataclass` ¶

RebalanceSpec `dataclass` ¶

from_horizon `staticmethod` ¶

Cadence `dataclass` ¶