Skip to content
betting-combat

betting_combat.jobs.rounds

Strategy 4 (rounds): its data, datasets, models and card runs as Dagster flows (ROUND_SYSTEM_FLOW §6).

RoundsContext  what every step needs: the databases, its ``data_jobs`` row, the run's card
               date, the one way to record what it landed, and the launch gate. Every flow
               of this location subclasses it (the settlement flow too).
DataFlow       raw landings only: Kalshi (the six series' markets, every minute candle and
               trade of each over its whole life, the winner markets' hourly candles) and
               UFC (the UFCStats mirror, Wikipedia's event list and card pages,
               MMADecisions' judge scores)
DatasetsFlow   fights_kalshi (each settled fight's card-start price), then the dataset
               tables the refit trains on (``services.rounds_datasets``)
ModelsFlow     the refit locked to the next card (promoted by its gate), the forward check
               of the traded cards, and the health watch (freshness, edge CUSUM)
TradingFlow    the card run (``trading.session.CardSession``) in paper (or shadow) or live

Every step reads its data_jobs row first (a disabled row is a named skip, before any vendor call), calls one service, lands through the store and records what it landed (a target receipt per table, grouped findings). Every row is seeded disabled; every sensor is born stopped and asks its job’s rows first (enabled_for_job: no row enabled, no launch). No job has a cron: the UFC calendar triggers them (services.rounds_calendar):

rounds_calendar_sensor         at most one chain per tick (``calendar_tick``), a prep before
                               a forward check: from 18:00 UTC two days before a card Kalshi
                               lists until its first fight - lead - 3 h, retried every tick,
                               an alert past it -> rounds_data_update (prep chain); two days
                               after a traded card -> rounds_data_update (forward chain)
rounds_datasets_after_data     a chain's data update succeeded -> rounds_datasets_update
rounds_refit_after_datasets    a prep chain's datasets succeeded -> rounds_models_refit
                               (locked to the chain's card)
rounds_forward_after_datasets  a forward chain's datasets succeeded -> rounds_models_forward
rounds_trade_paper_sensor      first fight - card_lead_minutes, mode paper or shadow
rounds_trade_live_sensor       first fight - card_lead_minutes, mode live

Runs keep apart where it matters: the data steps hold the Postgres advisory lock rounds_data (a data step that finds it held is a named skip); the datasets steps and the refit’s reads wait for it, the refit and the forward check for rounds_models, up to 30 min, then FAIL (a collision delays them, never drops them silently); the calendar sensor starts no chain while another chain, any rounds data/datasets/models run or any card run is under way; there is one card run per card across both trade jobs, a trade run refuses to start beside another run of its card, and it refuses an active model not locked to its card.

Classes

DataFlow

Bases: RoundsContext

Raw landings only. Each is idempotent (an upsert on the natural key, or an atomic swap of a whole snapshot) and resumable (a market is pulled once and recorded in the pull ledger). rounds_data_build takes everything not yet pulled or scraped (hours of paced requests); rounds_data_update the markets settled, the events dated and the decisions published in its rows’ lookback window (14 days: every card since the last chain, which runs before and after every card).

land_card_segments

land_card_segments() -> None

Wikipedia’s event list, then the card segments of the event pages dated in the window (a build: every past page not yet scraped).

land_judge_scores

land_judge_scores() -> None

MMADecisions: every decision not yet stored, this year’s (a build: from 2005).

land_kalshi_hourly

land_kalshi_hourly() -> None

The hourly candles of each settled fight’s two winner markets, the day after its card (the research’s 48 hours to the close): fights_kalshi’s input.

land_kalshi_markets

land_kalshi_markets() -> None

The six series’ markets (archive and settled), then every minute candle of every settled market over its whole life (only what its stored candles do not cover yet).

land_kalshi_trades

land_kalshi_trades() -> None

Every trade of every settled market over its whole life (only what its stored trades do not cover yet).

land_ufcstats

land_ufcstats() -> None

The whole UFCStats mirror (events, results, round stats, fighters) as one snapshot.

rounds_calendar_sensor

rounds_calendar_sensor(context: dg.SensorEvaluationContext)

The UFC calendar (services.rounds_calendar): at most one chain per tick, a prep (the card’s data update -> datasets -> refit) before a forward check (data update -> datasets -> forward check). An overdue prep is an alert (an error in the log).

DatasetsFlow

Bases: RoundsContext

The derived inputs and the dataset tables (services.rounds_datasets), from the raw tables only, under the data lock (nothing lands while they are read). The build rebuilds every fight’s frames and refits every clock year; the update reuses those whose inputs did not change.

build_round_datasets

build_round_datasets() -> None

Dataset 1 and 2, the clock and its markets, Set 4, Set 3 and the fills, from the raw tables: the update reuses every fight’s frames and every year’s clock fit whose inputs did not change, the build rebuilds all of them.

derive_fights_kalshi

derive_fights_kalshi() -> None

Each settled fight’s Kalshi card-start price (research ufc/pull.py) from its two winner markets’ hourly candles, the day after its card; a row is never rewritten. The datasets build after it needs it: with the build’s row on, a disabled row here FAILS the run (a skip would skip the build too, and the run would report success).

rounds_datasets_after_data

rounds_datasets_after_data(context: dg.SensorEvaluationContext)

A prep or forward chain’s data update succeeded: its datasets update.

ModelsFlow

Bases: RoundsContext

The refit (every model locked on the cards before the next card, registered as a candidate and promoted when strategies.auto_promote and its gate allow: services.rounds_refit), the forward check (each traded card replayed with the version it traded: services.rounds_forward) and the health watch (data freshness and the edge CUSUM per mode: services.rounds_health). Health runs in both jobs: the night before a card it shows whether the inputs are fresh, after a card whether the edge holds.

check_rounds_forward

check_rounds_forward() -> None

Each settled card of the window with live decisions, not checked yet.

refit_rounds_models

refit_rounds_models() -> None

Lock every model on the cards before the chain’s card (the launchpad’s or the betting/card_date tag; none: the next card) and land a candidate version.

rounds_forward_after_datasets

rounds_forward_after_datasets(context: dg.SensorEvaluationContext)

A forward chain’s datasets update succeeded: the forward check.

rounds_refit_after_datasets

rounds_refit_after_datasets(context: dg.SensorEvaluationContext)

A prep chain’s datasets update succeeded: the refit to the next card.

watch_rounds_health

watch_rounds_health() -> None

Data freshness and the edge CUSUM.

PaperOnlyConfigs

Bases: ConfigStore

The card run’s config reads in a paper run: the rounds row set to live is refused with ConfigError. At the start the run fails (the live sensor then launches the live job); mid-card the session keeps its last good configs and records the refusal (CardSession._refresh_configs). A paper run never sends a real order.

rounds

rounds() -> RoundsConfig | None

RoundsContext

Bases: DGFlow

What every step needs before it runs: the databases, its config row, the run’s card date, the one way to record what it landed, and the launch gate.

config_row

config_row(build: str, update: str) -> DataJobConfig

This step’s data_jobs row (the build row in a build job). A disabled row ends the step as a named skip before any vendor call.

data_health

data_health(surface: str, rows: int, latest_at: Any, notes: str | None = None) -> None

One Data & Jobs dashboard row; a monitoring failure never fails the step.

enabled_for_job

enabled_for_job(job_name: str) -> list[str]

The kill switch Dagster asks before a sensor launches a job.

Returns the enabled data_jobs rows among the job’s steps; an empty list means the job does not launch. Each step names its rows in @step(tags={"row.build": ..., "row.update": ...}).

exclusive

exclusive(lock: str, wait_s: float = 0.0, held: str) -> AsyncIterator[None]

Hold the advisory lock lock for the block; when another run holds it (after wait_s seconds), the step ends as a named skip: held.

job_kind

job_kind() -> str

Dagster tags every run with the kind of job that launched it: ‘build’ or ‘update’.

record

record(table: str, rows_written: int, problems: Counter[str] | None = None, rows_total_after: int | None = None, database: str | None = None) -> None

One target receipt for table (partial when anything was unfulfilled), grouped findings, and the receipt handed to the product’s checks.

run_date

run_date() -> dt.date

The card this run is about (the launchpad’s, else the sensor’s tag), else today in New York (cards are dated there; the UTC date is already tomorrow after 8 pm).

setup

setup(cfg: RunConfig) -> None

since

since(lookback_days: int | None) -> dt.datetime | None

An update’s window start (its row’s lookback); a build, or no lookback: everything.

teardown

teardown() -> None

waiting

waiting(lock: str, held: str, wait_s: float | None = None) -> AsyncIterator[bool]

Hold the advisory lock lock for the block, waiting up to wait_s seconds (default LOCK_WAIT_S, 30 min) for another run to let it go; still held then, the step FAILS (held): a collision delays a datasets or models step, never drops it silently.

Tick

What the calendar sensor does this tick: launch chain (its data update), or skip with why; alert: an overdue prep, logged as an error.

TradingFlow

Bases: RoundsContext

One UFC card, start to finish, in one long step: the existing card run (trading.session.CardSession), in the mode of the strategies row rounds. The paper job runs paper and shadow (neither sends a real order); the live job runs live. A job refuses (fails, nothing traded) when the row’s mode is not its own, when it is the paper job and any strategy is live, or when another run of the card is under way. A run that dies is relaunched by its sensor under a new run key and resumes from the database (fight clocks, decisions, orders, the pool): it never repeats a decision or an order.

rounds_trade_live_sensor

rounds_trade_live_sensor(context: dg.SensorEvaluationContext)

The card run in live mode, card_lead_minutes before the first fight.

rounds_trade_paper_sensor

rounds_trade_paper_sensor(context: dg.SensorEvaluationContext)

The card run in paper (or shadow) mode, card_lead_minutes before the first fight.

trade_rounds_live

trade_rounds_live() -> None

One card session in live mode: real orders on Kalshi.

trade_rounds_paper

trade_rounds_paper() -> None

One card session in paper (or shadow) mode: decide every break, fill against the books a second later (shadow: record only), settle, record.

Functions

busy

busy(instance: dg.DagsterInstance, now: dt.datetime) -> str | None

Why no chain may start now (a rounds run under way, or a chain waiting for its next stage), or None.

calendar_tick

calendar_tick(instance: dg.DagsterInstance, now: dt.datetime, listed: Mapping[dt.date, dt.datetime | None], portfolio: Any, lookback_days: int, locked: set[dt.date], decided: set[dt.date]) -> Tick

One tick of the calendar (see services.rounds_calendar). listed: Kalshi’s cards and their first fights; locked: the cards a model version is locked to; decided: the cards with rounds decisions recorded.

Prep first: the first listed card whose prep is due and not done (a version locked to it, or a prep chain of it whose refit succeeded), as a new chain (prep_chain) after every failed one, up to PREP_ATTEMPTS; while any rounds run or chain is under way it waits. Then the forward check: the first traded card due whose forward chain never ran (by its tag, whenever it ran), skipping any card already launched.

follow

follow(flow: DGFlow, instance: dg.DagsterInstance, job: str) -> dg.RunRequest | dg.SkipReason

The next stage of the chain waiting for job (the oldest such chain), or a skip.

one_row

one_row(name: str) -> dict[str, str]

A step with one data_jobs row, whichever job runs it.

rows

rows(what: str) -> dict[str, str]

A step’s two data_jobs rows: build_rounds_<what> and update_rounds_<what>.

run_views

run_views(instance: dg.DagsterInstance, jobs: Iterable[str], since: dt.datetime | None = None, tags: Mapping[str, str] | None = None) -> list[RunView]

The runs of jobs created after since (with tags), oldest first.

trade_job_for

trade_job_for(mode: Mode) -> str | None

The trade job of a rounds mode: live -> the live job; paper or shadow (neither sends a real order) -> the paper job; off -> none.

trade_refusal

trade_refusal(job: str, rounds: RoundsConfig | None) -> str | None

Why job may not run the card now (None: it may): the rounds strategy must trade in the job’s mode.

version_card

version_card(metrics: Mapping[str, Any]) -> dt.date | None

The card a rounds model version is locked to (its card_date; older: its cutoff).