Strategy Examples

These are intentionally small examples meant to get you from blank editor to a runnable backtest. They are educational starting points, not production strategies.

Example 1: Subscribe to one symbol and observe bars

Use this when you want the smallest possible strategy that validates, subscribes to one bar stream, and logs what it sees without placing orders.

strategy.py

from datetime import datetime

from libs.shared_ipc.src.framework import (
    DataCoordinator,
    StrategyBase,
    require_symbol,
    validate_config,
)
from libs.shared_ipc.src.protocol import (
    Action,
    BarEvent,
    BarRegistrationRequest,
    BarTimeSpan,
    RegisterSecurityAction,
)
from libs.common_types.src.types import JSONValue


class Strategy(StrategyBase):
    def __init__(
        self,
        data_coordinator: DataCoordinator,
        config: dict[str, JSONValue],
    ) -> None:
        validate_config(config)
        self._symbol = require_symbol(config, "symbol")

    async def on_start(self, as_of_time: datetime) -> list[Action]:
        return [
            RegisterSecurityAction(
                request=BarRegistrationRequest(
                    symbol=self._symbol.symbol,
                    asset_type=self._symbol.asset_type,
                    bar_time_span=BarTimeSpan.MINUTE,
                ),
            ),
        ]

    async def on_bar(self, event: BarEvent) -> list[Action]:
        print(
            f"{event.simulation_time.isoformat()} "
            f"{event.symbol} close={event.bar.close}"
        )
        return []

config_schema.json

{
  "type": "object",
  "properties": {
    "symbol": {
      "x-periscope-type": "symbol",
      "x-periscope-asset-type": "EQUITY",
      "type": "string",
      "description": "Equity symbol, e.g. SPY."
    }
  },
  "required": ["symbol"]
}

Example 2: Two moving averages on one equity

This is the first example that actually trades. It keeps two rolling windows on the same symbol and buys when the short average moves above the long average, then closes when it moves back below.

strategy.py

from collections import deque
from datetime import datetime
from decimal import Decimal

from libs.shared_ipc.src.framework import (
    DataCoordinator,
    StrategyBase,
    require_int,
    require_str,
    validate_config,
)
from libs.shared_ipc.src.protocol import (
    Action,
    AssetType,
    BarEvent,
    BarRegistrationRequest,
    BarTimeSpan,
    InsertCashAction,
    OrderIntent,
    PlaceOrderAction,
    RegisterSecurityAction,
)
from libs.common_types.src.types import JSONValue


class Strategy(StrategyBase):
    def __init__(
        self,
        data_coordinator: DataCoordinator,
        config: dict[str, JSONValue],
    ) -> None:
        validate_config(config)
        self._data_coordinator = data_coordinator
        self._symbol = require_str(config, "symbol")
        self._short_window = require_int(config, "short_window")
        self._long_window = require_int(config, "long_window")
        self._shares = require_int(config, "shares")
        self._closes: deque[Decimal] = deque(maxlen=self._long_window)

    async def on_start(self, as_of_time: datetime) -> list[Action]:
        return [
            InsertCashAction(amount=Decimal("100000")),
            RegisterSecurityAction(
                request=BarRegistrationRequest(
                    symbol=self._symbol,
                    asset_type=AssetType.EQUITY,
                    bar_time_span=BarTimeSpan.MINUTE,
                ),
            ),
        ]

    async def on_bar(self, event: BarEvent) -> list[Action]:
        if event.symbol != self._symbol or event.asset_type != AssetType.EQUITY:
            return []

        self._closes.append(event.bar.close)
        if len(self._closes) < self._long_window:
            return []

        short_values = list(self._closes)[-self._short_window:]
        short_ma = sum(short_values) / Decimal(len(short_values))
        long_ma = sum(self._closes) / Decimal(len(self._closes))

        position = await self._data_coordinator.get_position_by_symbol(
            self._symbol,
            AssetType.EQUITY,
        )

        if short_ma > long_ma and position is None:
            return [
                PlaceOrderAction(
                    symbol=self._symbol,
                    asset_type=AssetType.EQUITY,
                    intent=OrderIntent.BUY,
                    contract_quantity=Decimal(self._shares),
                ),
            ]

        if short_ma < long_ma and position is not None:
            return [
                PlaceOrderAction(
                    symbol=self._symbol,
                    asset_type=AssetType.EQUITY,
                    intent=OrderIntent.CLOSE,
                ),
            ]

        return []

config_schema.json

{
  "type": "object",
  "properties": {
    "symbol": {
      "type": "string",
      "description": "Equity symbol, e.g. SPY."
    },
    "short_window": {
      "type": "integer",
      "description": "Bars used in the short moving average."
    },
    "long_window": {
      "type": "integer",
      "description": "Bars used in the long moving average."
    },
    "shares": {
      "type": "integer",
      "description": "Shares to buy on each entry."
    }
  },
  "required": ["symbol", "short_window", "long_window", "shares"]
}

Example 3: Buy on a timer

This shows the other common pattern: use config to schedule recurring work and place orders from on_timer instead of from each bar. The bar subscription stays in place so the sim has pricing for the symbol; the on_bar handler itself does nothing.

strategy.py

from datetime import datetime
from decimal import Decimal

from libs.shared_ipc.src.framework import (
    DataCoordinator,
    StrategyBase,
    require_decimal,
    require_int,
    require_str,
    validate_config,
)
from libs.shared_ipc.src.protocol import (
    Action,
    AssetType,
    BarEvent,
    BarRegistrationRequest,
    BarTimeSpan,
    InsertCashAction,
    OrderIntent,
    PlaceOrderAction,
    RegisterSecurityAction,
    RegisterTimeStepAction,
    TimerEvent,
)
from libs.common_types.src.types import JSONValue


class Strategy(StrategyBase):
    def __init__(
        self,
        data_coordinator: DataCoordinator,
        config: dict[str, JSONValue],
    ) -> None:
        validate_config(config)
        self._symbol = require_str(config, "symbol")
        self._tick_seconds = require_int(config, "tick_seconds")
        self._shares_per_tick = require_decimal(config, "shares_per_tick")

    async def on_start(self, as_of_time: datetime) -> list[Action]:
        return [
            InsertCashAction(amount=Decimal("100000")),
            RegisterSecurityAction(
                request=BarRegistrationRequest(
                    symbol=self._symbol,
                    asset_type=AssetType.EQUITY,
                    bar_time_span=BarTimeSpan.MINUTE,
                ),
            ),
            RegisterTimeStepAction(time_step_seconds=float(self._tick_seconds)),
        ]

    async def on_bar(self, event: BarEvent) -> list[Action]:
        return []

    async def on_timer(self, event: TimerEvent) -> list[Action]:
        return [
            PlaceOrderAction(
                symbol=self._symbol,
                asset_type=AssetType.EQUITY,
                intent=OrderIntent.BUY,
                contract_quantity=self._shares_per_tick,
            ),
        ]

config_schema.json

{
  "type": "object",
  "properties": {
    "symbol": {
      "type": "string",
      "description": "Equity symbol, e.g. SPY."
    },
    "tick_seconds": {
      "type": "integer",
      "description": "Seconds between scheduled purchases."
    },
    "shares_per_tick": {
      "type": "number",
      "description": "Shares to buy on each tick."
    }
  },
  "required": ["symbol", "tick_seconds", "shares_per_tick"]
}

What to change next

• Swap the symbol and window sizes in the moving-average example and rerun the same backtest date range to see how sensitivity changes.
• Add a second RegisterSecurityAction in on_start and filter by event.symbol in on_bar to trade more than one instrument from one strategy.
• Replace the fixed-size order in the timer example with a cash-proportion target, reading target_allocation as a new number field on the schema.