Declare it once. Backtest it honestly.

From the SDK

Ember-bound dogfood (ADR 0011, end to end)

python dogfood_ember.py               # local, prints the comparison
MODE=remote python dogfood_ember.py   # submit BOTH runs to forge

examples/dogfood_ember.py · 233 linesfrom __future__ import annotations

import os
from datetime import datetime, timedelta, timezone
from typing import Any, Mapping, TypedDict

from promethean import backtest as bt
from promethean.backtest import BacktestContext, Order
from promethean.pipelines import pipeline, step
from promethean.strategy import Feed, backtest as realism, strategy

SYMBOL = "SYNTH_A"
TICK = 0.01
START = datetime(2025, 1, 2, tzinfo=timezone.utc)
DAYS = 300  # walk past the last row; PIT reads clamp to what existed


# ---------------------------------------------------------------------------
# 1. Events from Ember, point-in-time
# ---------------------------------------------------------------------------

@step(name="events")
def fetch_events() -> list[dict]:
    """One as-of read per day: the book the strategy sees is what Ember knew."""
    # Imported HERE, not at module level. A module-level import would put the
    # PyO3 `Client` class in this step's captured globals, and a step is
    # pickled by value (a pipeline file is a script, so it is not importable on
    # the worker). Resolving it at call time keeps it out of the code blob.
    #
    # Since 0.3.21 that is good practice rather than a load-bearing workaround:
    # the native classes declare `module = "promethean._native"` so they pickle
    # by reference, and `submit` refuses a blob that still cannot be
    # deserialized instead of letting the run fail remotely (BE-272). Written
    # this way because it is the habit worth copying out of this file.
    from promethean import Client

    # Unreachable store and empty store are different mistakes with different
    # fixes, so they get different messages. The raw failure here is a bare
    # "Connection error: transport error", which says nothing about what this
    # example needs.
    try:
        c = Client()
        rows, seen = [], set()
        for d in range(DAYS):
            asof = (START + timedelta(days=d)).isoformat().replace("+00:00", "Z")
            r = c.lookup("SpotPrices", {"symbol": SYMBOL}, timestamp=asof)
            if r and r["timestamp"] not in seen:
                seen.add(r["timestamp"])
                rows.append(r)
    except Exception as e:
        raise RuntimeError(
            f"cannot read SpotPrices from Ember ({e}). This example is the one "
            f"that trades *real store data*, so unlike the other examples here "
            f"it needs a running cluster — set EMBER_URL if you are outside a "
            f"workspace pod. See this module's docstring for the full "
            f"prerequisites."
        ) from e

    rows.sort(key=lambda r: r["timestamp"])
    if not rows:
        raise RuntimeError(f"no {SYMBOL} rows in Ember — seed with spot_price_pipeline.py first")

    events = []
    for r in rows:
        ts = r["timestamp"]
        iso = ts.isoformat().replace("+00:00", "Z")
        px = round(float(r["close"]), 2)
        half = max(TICK, round(px * 0.0005, 2))
        depth = max(1.0, r["volume"] / 2e6)
        bids = [[round(px - half - i * TICK, 2), depth * (1 - 0.15 * i)] for i in range(5)]
        asks = [[round(px + half + i * TICK, 2), depth * (1 - 0.15 * i)] for i in range(5)]
        events.append({"ts": iso, "kind": {"Snapshot": {"bids": bids, "asks": asks}}})
        iso_t = (ts + timedelta(milliseconds=500)).isoformat().replace("+00:00", "Z")
        side = "Buy" if float(r["close"]) >= float(r["open"]) else "Sell"
        events.append({"ts": iso_t, "kind": {"Trade": {
            "price": px + half if side == "Buy" else px - half,
            "qty": round(max(0.5, r["volume"] / 4e7), 2),
            "aggressor": side,
        }}})
    print(f"[fetch] {len(rows)} PIT rows -> {len(events)} market events")
    return events


# ---------------------------------------------------------------------------
# 2. The strategy — ordinary Python, daily momentum with a resting exit
# ---------------------------------------------------------------------------

def _mk_strategy():
    @strategy(
        name="ember-momentum",
        subscriptions=[Feed("SpotPrices", keys=[SYMBOL], resolution={"interval": 86400})],
    )
    class EmberMomentum:
        """Long when close > SMA(5); resting take-profit 0.8% above entry."""

        LOOKBACK, TP_BPS, CLIP = 5, 80, 2.0

        def __init__(self):
            self.closes, self.next_id, self.tp_id = [], 1, None

        def on_event(self, ctx: BacktestContext):
            if ctx.mid is None:
                return None
            self.closes.append(ctx.mid)
            if len(self.closes) <= self.LOOKBACK:
                return None
            sma = sum(self.closes[-self.LOOKBACK:]) / self.LOOKBACK
            out = []
            if ctx.position == 0 and ctx.mid > sma:
                out.append(Order.market("buy", self.CLIP))
                tp = round(ctx.mid * (1 + self.TP_BPS / 10_000), 2)
                self.tp_id = self.next_id
                self.next_id += 1
                out.append(Order.limit(self.tp_id, "sell", tp, self.CLIP))
            elif ctx.position > 0 and ctx.mid < sma:
                if self.tp_id is not None:
                    out.append(Order.cancel(self.tp_id))
                    self.tp_id = None
                out.append(Order.market("sell", ctx.position))
            return out or None

    return EmberMomentum


# ---------------------------------------------------------------------------
# 3. Two runs: idealized vs declared realism
# ---------------------------------------------------------------------------

class RealismConfig(TypedDict, total=False):
    """The `@backtest(...)` knobs, as a type so a config constant can be one.

    A realism config is a heterogeneous mapping — `venue` is a string, `seed` an
    int, `fees` a dict of floats — so declaring it as a plain `dict` collapses
    the values to a union and every `**REALISM` unpack becomes a type error at
    the decorator. Spelling the shape out is how a config constant stays
    checkable: misspell a knob or hand `partials` a float and you hear about it
    here, not three hours into a run.

    `total=False` because the whole point of the sparse dict is that an unset
    knob inherits the venue default rather than an SDK-side guess. Mirrors
    `promethean.strategy.backtest`'s signature.
    """

    venue: str
    latency: Mapping[str, Any]
    queue: str
    fees: Mapping[str, Any]
    slippage: Mapping[str, Any]
    partials: str
    seed: int


# A dict *literal*, not `dict(...)`: a TypedDict is only inferred from the
# literal form, and the whole point of annotating it is to be told when a knob
# is misspelled or mistyped.
REALISM: RealismConfig = {
    # Start from a shipped venue profile, then say only what differs (BE-271).
    # Precedence is: declared field > venue default > idealized, resolved
    # engine-side — so the fee ladder, lot grid and feed leg below come from
    # Coinbase's profile without being spelled out here. Read what that buys
    # with `promethean.realism.defaults("coinbase")`, and check its
    # `_provenance`: today these are priors from published venue docs, not fits
    # to our own fills.
    "venue": "coinbase",
    "latency": {"submit": 5000.0},   # on a daily grid: you act on the NEXT bar
    "queue": "size_aware",
    "fees": {"maker_bps": 4.0, "taker_bps": 20.0},
    "slippage": {"square_root": 1.5},
    "partials": "rest",
    "seed": 11,
}


@step(name="backtest-ideal", depends_on=["events"])
def backtest_ideal(events: list[dict]) -> dict:
    res = bt.run_lob(_mk_strategy()(), events, tick_size=TICK)
    print(f"[ideal]   fills={res.num_trades} return={res.total_return:+.4%} "
          f"sharpe={res.sharpe:.3f} maxdd={res.max_drawdown:.4%}")
    return {"fills": res.num_trades, "ret": res.total_return, "sharpe": res.sharpe}


@step(name="backtest-realism", depends_on=["events"])
def backtest_realism(events: list[dict]) -> dict:
    cls = realism(**REALISM)(_mk_strategy())
    res = bt.run_lob(cls(), events, tick_size=TICK)
    print(f"[realism] fills={res.num_trades} return={res.total_return:+.4%} "
          f"sharpe={res.sharpe:.3f} maxdd={res.max_drawdown:.4%}")
    return {"fills": res.num_trades, "ret": res.total_return, "sharpe": res.sharpe}


@step(name="report")
def report(ideal: dict, real: dict) -> dict:
    gap = ideal["ret"] - real["ret"]
    print("=" * 56)
    print(f"idealized : fills={ideal['fills']:>4} return={ideal['ret']:+.4%} sharpe={ideal['sharpe']:.3f}")
    print(f"realistic : fills={real['fills']:>4} return={real['ret']:+.4%} sharpe={real['sharpe']:.3f}")
    print(f"realism cost: {gap:+.4%} of return — the number the idealized run hides")
    print("=" * 56)
    return {"ideal": ideal, "realism": real, "realism_cost": gap}


# ---------------------------------------------------------------------------
# Pipelines — one per mode so each gets its own Backtest view
# ---------------------------------------------------------------------------

@pipeline(name="dogfood-ember-ideal", version=1)
def dogfood_ideal():
    return [fetch_events, backtest_ideal]


@pipeline(name="dogfood-ember-realism", version=1)
def dogfood_realism():
    return [fetch_events, backtest_realism]


def _run_local():
    ev = fetch_events()
    report(backtest_ideal(ev), backtest_realism(ev))


def _submit_remote():
    from promethean.pipelines.submit import submit
    for fn in (dogfood_ideal, dogfood_realism):
        run_id = submit(None, pipeline_fn=fn)
        print(f"submitted {fn.pipeline_def['name']}: {run_id}")


if __name__ == "__main__":
    if os.environ.get("MODE") == "remote":
        _submit_remote()
    else:
        _run_local()