Architecture

Overview

The data flow is linear and unidirectional: a Provider fetches raw data, the Agent evaluates it and emits a BetIntent, and the Executor converts that intent into a Position. All layers communicate through the Core models: pure data classes with no I/O.

Agent (strategy)
  │ BetIntent
  ▼
Executor (Simulator / Exchange)
  │ Position
  ▼
Provider (Events + Odds)
  │
  ▼
Core models (Event · OddsSnapshot · Position · ...)

Layer 1: Core models

Located in python/opensport/core/ and typescript/src/core/. These are immutable data classes with no I/O. All layers depend on them.

Odds format

Opensport stores all odds in decimal format internally. Conversion helpers are provided for American (moneyline) and fractional formats. Implied probability is always 1 / decimal_odds.

Layer 2: Providers

A Provider fetches events and odds from a data source and normalises them into the core models. Implement BaseProvider to connect any feed.

class BaseProvider(ABC):
    def get_events(sport=None, competition=None, status=None) -> list[Event]: ...
    def get_odds(event_id: str) -> OddsSnapshot: ...
    # Optional:
    def get_markets(event_id: str) -> list[Market]: ...
    def get_live_score(event_id: str) -> dict: ...
    async def stream_odds(event_id: str, interval_seconds=5.0): ...

Managing multiple providers

Use ProviderRegistry to track which providers are active and MultiProvider to expose all of them through a single BaseProvider interface. Because MultiProvider is a BaseProvider, agent code requires no changes. See the Providers reference for full examples and constructor options.

Building a custom provider

Subclass BaseProvider, implement get_events() and get_odds(), and map your data source's native response objects into the core models. See the Providers reference for a full skeleton.

Layer 3: Execution

An Executor accepts a BetIntent and returns a Position. The same agent code runs against the Simulator (paper trading) or a live exchange. Just swap the executor.

class BaseExecutor(ABC):
    def place(intent: BetIntent) -> Position: ...
    def cancel(position_id: str) -> bool: ...
    def get_position(position_id: str) -> Position: ...
    def get_balance() -> float: ...

Risk guards in ExchangeExecutor

• max_single_stake: hard cap on any single bet
• max_total_exposure: abort if aggregate open stakes exceed limit
• dry_run=True: log intended actions without calling the exchange API

Layer 4: Agents

An Agent orchestrates the evaluate → place loop. Subclass BaseAgent and implement evaluate(). The base class handles fetching, stake clamping, open-position limits, and error recovery.

class BaseAgent(ABC):
    def evaluate(event: Event, snapshot: OddsSnapshot) -> BetIntent | None: ...
    def run(sport=None, limit=None) -> list[Position]: ...

Built-in agents

End-to-end data flow

1. provider.get_events(sport="soccer")
      → [Event(Arsenal vs Chelsea, scheduled), ...]

2. provider.get_odds("mock_soccer_000")
      → OddsSnapshot(
          markets=[
            MarketOdds("winner",  [Home@2.10, Draw@3.40, Away@4.20]),
            MarketOdds("totals",  [Over2.5@1.85, Under2.5@1.95]),
            MarketOdds("btts",    [Yes@1.80, No@2.05]),
          ]
        )

3. agent.evaluate(event, snapshot)
      → BetIntent(BACK "Home" @ ≥2.10, stake=47.50)

4. executor.place(intent)
      → Position(id="a3f9b2c1", status=PENDING, oddsTaken=2.10, stake=47.50)

5. (after match) executor.settle_position(pos.id, won=True)
      → Position(status=WON, settlement_value=+52.50)

Testing strategy

Every layer can be tested in isolation using MockProvider (deterministic via seed) and Simulator (in-memory, zero I/O). No API keys or network access are required for the full test suite.

provider = MockProvider(seed=42)   # deterministic
sim      = Simulator(bankroll=1000, verbose=False)
agent    = MyAgent(provider=provider, execution=sim)
positions = agent.run(limit=5)
assert sim.get_balance() >= 0