Kalshi Basket Arbitrage Backtester

Build a constraint-aware backtester for prediction-market basket arbitrage.

Introduction

30 Second Summary

A group of bets can look like free money when its prices add up to less than one dollar. One overlooked outcome can erase that apparent profit.

In this project, you will build a local prediction-market arbitrage research engine with Python, SciPy, and Matplotlib. It exposes a false signal before using an explicit payoff matrix to turn synthetic quote snapshots into CSV decisions plus a PNG analytics chart.

What You'll Build

You will run the finished engine to see each snapshot classified with its edge, capacity, and cumulative hypothetical profit.

By the end of this project, you'll have:

  • A runnable false-signal test that reports a naive edge of $0.0700 while the none-of-the-listed-outcomes state pays $0.0000.
  • A constraint-aware optimizer that uses linear programming to select YES and NO contract fractions. Its chosen basket maintains a minimum payout of $1.0000 across every feasible state.
  • A replay backtest that classifies four snapshots under quoted depth plus an execution haircut. It exports auditable decisions to backtest_results.csv. Its chart tracks gross edge plus cumulative hypothetical profit in arb_report.png.
  • Secret Mission: An authenticated Kalshi Demo collector reconstructs top-of-book asks. The same engine replays its captured snapshot without placing orders.

Are there any prerequisites?

This project assumes you are comfortable reading Python functions, JSON, and basic linear inequalities. The main path uses local synthetic data, so Kalshi Demo credentials plus an RSA private key are only needed for the optional Secret Mission.

Before We Start

Before any hands-on work, lock in the standard that every later result must meet. Your backtester will minimize basket cost while requiring at least $1.00 of payout in every explicitly modeled feasible settlement state. It remains a research tool that never submits order requests.

Set Up the Research Workspace

The backtester depends on Python, SciPy, and Matplotlib producing consistent results. A reproducible environment keeps the optimizer and charts stable while you build.

This step gives the project its own isolated virtual environment. A version-pinned requirements file records the exact packages behind every result.

In this step, get ready to:
  • Verify that Python 3.12 or newer is available.
  • Create an isolated workspace inside kalshi-basket-backtester.
  • Install SciPy 1.18.1 plus Matplotlib 3.11.2 from requirements.txt.
Verify Python and PyCharm

SciPy 1.18.1 requires Python 3.12 or newer. Checking the runtime first prevents an incompatible package installation later.

  • Press Cmd+Space to open macOS search.
  • Type Terminal into the search bar. Press Enter to open macOS Terminal.
  • Check your installed Python version by running this command:
python3 --version

What does this check prove?

The command reports the Python runtime attached to python3. A version of 3.12 or newer can install the pinned SciPy release.

✔️ I see Python 3.12 or higher

Your Python runtime meets the project requirement.

  • Keep Terminal open for the workspace commands below.

ⓧ I see an older version

Your current runtime cannot use SciPy 1.18.1. The current stable macOS release is Python 3.14.7.

  • Visit the official Python macOS downloads page.
  • Download the current stable macOS release.
  • Open the downloaded installer.
  • Complete the installation with the default options.
  • Return to Terminal.
  • Check the updated runtime by running this command:
python3 --version

What does this second check prove?

The version output confirms whether python3 now points to the newer installation. Continue when the reported version is 3.12 or higher.

ⓧ Command not found

Your shell cannot find a Python installation through the python3 command. Installing the current stable macOS release adds the runtime this project needs.

  • Visit the official Python macOS downloads page.
  • Download Python 3.14.7 for macOS.
  • Open the downloaded installer.
  • Complete the installation with the default options.
  • Return to Terminal.
  • Confirm the installation by running this command:
python3 --version

What does this confirmation prove?

A reported version confirms that Terminal can now find python3. Continue when the version is 3.12 or higher.

You will use PyCharm to manage the project files. A quick launch check reveals whether the editor is ready.

  • Press Cmd+Space to open macOS search.
  • Type PyCharm into the search bar. Press Enter to launch it.

✔️ PyCharm opens

PyCharm is ready to open the workspace after you create it.

  • Leave the PyCharm window open.

ⓧ PyCharm does not appear

PyCharm needs a one-time installation before it can open the workspace.

  • Visit the official PyCharm installation guide.
  • Download the macOS disk image.
  • Open the downloaded disk image.
  • Drag PyCharm into Applications.
  • Press Cmd+Space to open macOS search.
  • Type PyCharm into the search bar. Press Enter to launch it.
  • Leave the PyCharm window open.
Create and open the workspace

The project needs a predictable location plus an isolated package directory. The Desktop keeps the workspace easy to find while .venv keeps its dependencies separate from the rest of your Mac.

  • Switch back to Terminal.
  • Move to your Desktop by running this command:
cd ~/Desktop

What does this command do?

The cd command changes Terminal's current location. The ~ shortcut represents your macOS home folder.

  • Create the kalshi-basket-backtester folder. Move Terminal into it by running these commands:
mkdir kalshi-basket-backtester
cd kalshi-basket-backtester

What do these commands do?

  • The mkdir command creates the project folder on your Desktop.
  • The cd command moves the Terminal session into that folder. Every project command now runs from the correct location.

Your Terminal prompt now points to the new project folder. That path confirms the workspace exists in the intended location.

  • Create the virtual environment. Activate it by running these commands:
python3 -m venv .venv
source .venv/bin/activate

What do these commands do?

  • The first command creates an isolated Python environment inside .venv.
  • The second command activates that environment in the current Terminal session. Package installations now stay inside this project.

The activated environment usually adds (.venv) near the start of your Terminal prompt. Your project now has its own Python package space.

  • Switch back to the PyCharm window from the earlier launch check.
  • Click Open on the Welcome Screen.
  • Select the kalshi-basket-backtester folder on your Desktop.
  • Click Open in the folder dialog.

You should see kalshi-basket-backtester at the top of PyCharm's Project tool window. The hidden .venv directory may also appear in the file tree.

Workspace missing in PyCharm?

  • Check that you selected the kalshi-basket-backtester folder itself.
  • Confirm that the selected folder is inside your Desktop directory.
  • Return to the Open action if PyCharm only shows its Welcome Screen.
  • Help me open my project folder in PyCharm.
Pin and install the dependencies

A requirements file turns package versions into part of the project record. Anyone rebuilding this workspace can install the same optimizer and plotting library.

  • Select kalshi-basket-backtester in PyCharm's Project tool window.
  • Click New in the tool window toolbar.
  • Select File from the creation options.
  • Enter requirements.txt as the file name.
  • Add the pinned dependencies by pasting this code:
scipy==1.18.1
matplotlib==3.11.2

What does this file control?

  • The SciPy pin provides the linear-programming optimizer used later in the backtester.
  • The Matplotlib pin provides the charting tools used for the final analytics report.
  • The exact version pins keep dependency behavior reproducible across installations.
  • Save requirements.txt.
  • Confirm that requirements.txt appears directly under kalshi-basket-backtester in the Project tool window.

File missing from the project?

  • Check that you selected the top-level kalshi-basket-backtester folder before creating the file.
  • Confirm that the filename is exactly requirements.txt.
  • Help me create requirements.txt in the correct PyCharm folder.

✔️ Awesome, I've got everything!

Great. Double check that you saved requirements.txt inside kalshi-basket-backtester.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference:

scipy==1.18.1
matplotlib==3.11.2

Both lines must match this reference exactly. The version pins determine which package releases the environment installs.

Before you install the file, which two package versions do you expect the command to use?

The first installation can take a few minutes because pip downloads the scientific packages.

  • Switch back to the activated Terminal session.
  • Install the pinned dependencies by running this command:
python3 -m pip install -r requirements.txt

What does this command do?

  • The python3 command uses the runtime from your activated virtual environment.
  • The pip module installs Python packages into that environment.
  • The -r option reads both package pins from requirements.txt.

You should see the installation finish without an error. The completion output identifies SciPy 1.18.1 plus Matplotlib 3.11.2.

Dependency installation failed?

  • Return to the Python version check if Terminal reports that a package is incompatible with your runtime.
  • Confirm that Terminal is still inside kalshi-basket-backtester if it cannot find requirements.txt.
  • Repeat the activation command from the workspace substep if the virtual environment is no longer active.
  • Help me debug my dependency installation.

That's your research workspace locked down. Next up, you will run a cheap-looking basket through a naive check and expose the settlement state it misses.

Trigger a False Arbitrage Signal

Your reproducible Python workspace is ready. You can now challenge a price signal without setup noise.

A basket priced below $1 can look profitable at first glance. That headline depends on which outcomes can settle together.

In this step, get ready to:
  • Model three feasible settlement states across six YES and NO legs.
  • Record four synchronized synthetic quote snapshots.
  • Run a naive basket check against every modeled state.
Model the feasible settlement states

A payoff matrix maps each feasible settlement state to each contract. Every value shows whether that contract pays $1 or $0 in that state.

The third state preserves the possibility that neither narrow outcome occurs. This state is essential for testing the price signal honestly.

  • Create constraints.json inside the open kalshi-basket-backtester folder by using the file creation option in the PyCharm project sidebar.

You'll see an empty constraints.json file listed beside requirements.txt.

  • Add the complete settlement model by using the full-file reference below:

✔️ Awesome, I've got everything!

Your constraints.json should contain three settlement states with the same six payoff legs.

ⓧ I'd like to double check the full code

{
  "basket_name": "Synthetic overlapping-outcomes basket",
  "naive_partition_legs": [
    "SYN-A:YES",
    "SYN-B:YES"
  ],
  "states": [
    {
      "name": "A occurs",
      "payouts": {
        "SYN-A:YES": 1,
        "SYN-A:NO": 0,
        "SYN-B:YES": 0,
        "SYN-B:NO": 1,
        "SYN-ANY:YES": 1,
        "SYN-ANY:NO": 0
      }
    },
    {
      "name": "B occurs",
      "payouts": {
        "SYN-A:YES": 0,
        "SYN-A:NO": 1,
        "SYN-B:YES": 1,
        "SYN-B:NO": 0,
        "SYN-ANY:YES": 1,
        "SYN-ANY:NO": 0
      }
    },
    {
      "name": "Neither A nor B occurs",
      "payouts": {
        "SYN-A:YES": 0,
        "SYN-A:NO": 1,
        "SYN-B:YES": 0,
        "SYN-B:NO": 1,
        "SYN-ANY:YES": 0,
        "SYN-ANY:NO": 1
      }
    }
  ]
}

How does this model work?

  • The naive_partition_legs list identifies the two YES contracts tested by the shortcut.
  • Each object in states represents one feasible settlement.
  • Each payouts object assigns a $1 or $0 payoff to all six legs.
  • The final state keeps the none-of-the-listed-outcomes possibility inside the model.
  • Save constraints.json.

You'll see matching braces around three state objects. PyCharm should not highlight any JSON syntax problems.

Is the JSON highlighted as invalid?

  • Check that every property except the final property in an object ends with a comma.
  • Confirm each payoff leg uses the same capitalization in all three states.
  • Compare the closing braces at the bottom of constraints.json with the full-file reference.
  • Help me find the JSON syntax problem.
Record synchronized quote snapshots

A JSONL file stores one complete JSON record per line. Each record captures quotes from one replay timestamp.

The synthetic records vary ask prices across time. Their depth values provide a quantity limit for each leg.

  • Create snapshots.jsonl inside the open kalshi-basket-backtester folder by using the file creation option in the project sidebar.

You'll see an empty snapshots.jsonl file in the project sidebar.

  • Add the four synthetic quote records by pasting this data:
{"timestamp":"2026-09-18T10:00:00Z","quotes":{"SYN-A:YES":{"ask":0.46,"depth":10},"SYN-A:NO":{"ask":0.56,"depth":10},"SYN-B:YES":{"ask":0.47,"depth":10},"SYN-B:NO":{"ask":0.55,"depth":10},"SYN-ANY:YES":{"ask":0.61,"depth":10},"SYN-ANY:NO":{"ask":0.42,"depth":10}}}
{"timestamp":"2026-09-18T10:01:00Z","quotes":{"SYN-A:YES":{"ask":0.42,"depth":10},"SYN-A:NO":{"ask":0.60,"depth":10},"SYN-B:YES":{"ask":0.43,"depth":10},"SYN-B:NO":{"ask":0.59,"depth":10},"SYN-ANY:YES":{"ask":0.83,"depth":10},"SYN-ANY:NO":{"ask":0.19,"depth":10}}}
{"timestamp":"2026-09-18T10:02:00Z","quotes":{"SYN-A:YES":{"ask":0.47,"depth":8},"SYN-A:NO":{"ask":0.54,"depth":8},"SYN-B:YES":{"ask":0.47,"depth":8},"SYN-B:NO":{"ask":0.54,"depth":8},"SYN-ANY:YES":{"ask":0.56,"depth":8},"SYN-ANY:NO":{"ask":0.45,"depth":8}}}
{"timestamp":"2026-09-18T10:03:00Z","quotes":{"SYN-A:YES":{"ask":0.43,"depth":5},"SYN-A:NO":{"ask":0.58,"depth":5},"SYN-B:YES":{"ask":0.43,"depth":5},"SYN-B:NO":{"ask":0.58,"depth":5},"SYN-ANY:YES":{"ask":0.80,"depth":5},"SYN-ANY:NO":{"ask":0.21,"depth":5}}}

What does this data represent?

  • Each timestamp anchors one replay record to a specific moment.
  • Each ask value represents the quoted cost of one contract.
  • Each depth value caps the available quantity for that quote.
  • The four records create changing market conditions for the later replay.
  • Save snapshots.jsonl.

You'll see exactly four non-empty lines in the editor. Every line should begin with a timestamp and contain quotes for all six legs.

Does a snapshot wrap or look incomplete?

  • Treat each visual wrap as part of the same record unless the editor shows a new line number.
  • Confirm the file contains four line numbers with data.
  • Check that every record ends with two closing braces.
  • Help me check my JSONL records.

✔️ Awesome, I've got everything!

Great. Double check that you saved all four records in snapshots.jsonl.

ⓧ I'd like to double check the full code

{"timestamp":"2026-09-18T10:00:00Z","quotes":{"SYN-A:YES":{"ask":0.46,"depth":10},"SYN-A:NO":{"ask":0.56,"depth":10},"SYN-B:YES":{"ask":0.47,"depth":10},"SYN-B:NO":{"ask":0.55,"depth":10},"SYN-ANY:YES":{"ask":0.61,"depth":10},"SYN-ANY:NO":{"ask":0.42,"depth":10}}}
{"timestamp":"2026-09-18T10:01:00Z","quotes":{"SYN-A:YES":{"ask":0.42,"depth":10},"SYN-A:NO":{"ask":0.60,"depth":10},"SYN-B:YES":{"ask":0.43,"depth":10},"SYN-B:NO":{"ask":0.59,"depth":10},"SYN-ANY:YES":{"ask":0.83,"depth":10},"SYN-ANY:NO":{"ask":0.19,"depth":10}}}
{"timestamp":"2026-09-18T10:02:00Z","quotes":{"SYN-A:YES":{"ask":0.47,"depth":8},"SYN-A:NO":{"ask":0.54,"depth":8},"SYN-B:YES":{"ask":0.47,"depth":8},"SYN-B:NO":{"ask":0.54,"depth":8},"SYN-ANY:YES":{"ask":0.56,"depth":8},"SYN-ANY:NO":{"ask":0.45,"depth":8}}}
{"timestamp":"2026-09-18T10:03:00Z","quotes":{"SYN-A:YES":{"ask":0.43,"depth":5},"SYN-A:NO":{"ask":0.58,"depth":5},"SYN-B:YES":{"ask":0.43,"depth":5},"SYN-B:NO":{"ask":0.58,"depth":5},"SYN-ANY:YES":{"ask":0.80,"depth":5},"SYN-ANY:NO":{"ask":0.21,"depth":5}}}

How should I compare this file?

Compare one record at a time. Each record must preserve its timestamp plus all six ask and depth pairs.

Run the naive basket check

The naive checker adds the asks for the two YES legs listed in naive_partition_legs. It then compares that headline price with the payout in every modeled state.

  • Create naive_check.py inside the open kalshi-basket-backtester folder by using the file creation option in the project sidebar.

You'll see an empty naive_check.py file beside the two data files.

  • Add the naive price check by pasting this code:
import json
from pathlib import Path

config = json.loads(Path("constraints.json").read_text(encoding="utf-8"))
first_line = Path("snapshots.jsonl").read_text(encoding="utf-8").splitlines()[0]
snapshot = json.loads(first_line)
legs = config["naive_partition_legs"]

cost = sum(float(snapshot["quotes"][leg]["ask"]) for leg in legs)
print(f"Naive basket cost: ${cost:.4f}")
print(f"Naive edge: ${1.0 - cost:.4f}")

for state in config["states"]:
    payout = sum(float(state["payouts"][leg]) for leg in legs)
    print(f"{state['name']}: payout=${payout:.4f}")

What does this code do?

  • The first section loads the settlement model from constraints.json.
  • The first_line variable selects the first quote snapshot.
  • The cost calculation adds the asks for the two naive partition legs.
  • The final loop calculates the basket payout separately for every feasible state.
  • Save naive_check.py.

PyCharm should show a complete script with no highlighted syntax problems. The final loop should remain outside the earlier expressions.

Is the script highlighted as invalid?

  • Check that the four lines under for state in config["states"]: use consistent indentation.
  • Confirm the dictionary keys retain their quotation marks.
  • Compare every closing parenthesis with the reference below.
  • Help me fix my naive checker.

✔️ Awesome, I've got everything!

Great. Double check that you saved naive_check.py.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

config = json.loads(Path("constraints.json").read_text(encoding="utf-8"))
first_line = Path("snapshots.jsonl").read_text(encoding="utf-8").splitlines()[0]
snapshot = json.loads(first_line)
legs = config["naive_partition_legs"]

cost = sum(float(snapshot["quotes"][leg]["ask"]) for leg in legs)
print(f"Naive basket cost: ${cost:.4f}")
print(f"Naive edge: ${1.0 - cost:.4f}")

for state in config["states"]:
    payout = sum(float(state["payouts"][leg]) for leg in legs)
    print(f"{state['name']}: payout=${payout:.4f}")

How should I compare this script?

Confirm the script loads both data files before calculating the cost. The state loop must check the same two legs selected by naive_partition_legs.

Before you run the checker, do you think a positive headline edge guarantees that every settlement state pays $1?

  • Test the naive basket from the activated terminal by running:
python3 naive_check.py

What should I see?

You'll see a naive basket cost of $0.9300. The headline calculation reports a naive edge of $0.0700.

The two named outcome states each pay $1.0000. The Neither A nor B occurs state pays $0.0000.

You caught the false signal. A positive headline edge does not prove a guaranteed payout across every feasible state.

Did the checker fail to run?

  • Confirm the terminal is still inside the kalshi-basket-backtester folder.
  • Verify that constraints.json and snapshots.jsonl sit beside naive_check.py.
  • Check both data files for missing commas or closing braces.
  • Help me debug the false-signal check.

Your synthetic basket now exposes the gap between a cheap price sum and a guaranteed payoff. Next, you'll turn every feasible settlement state into a hard optimization constraint.

Enforce Every Settlement State

Your naive checker exposed a positive headline edge that paid nothing when neither narrow outcome occurred. That result proved the settlement model must govern the basket.

In this step, you’ll use linear programming to make every feasible settlement state a hard constraint. The optimizer can only accept quantities whose lowest state payout reaches $1.00.

In this step, get ready to:
  • Load the existing constraint file and quote snapshots.
  • Build a payoff matrix for every feasible settlement state.
  • Use depth-bounded optimization to find the cheapest guaranteed basket.
Load and validate the inputs

The optimizer needs the same ordered leg list for every state. A consistency check stops a missing payoff column from silently shifting the payoff matrix.

  • Switch back to the project folder already open in PyCharm.
  • Create arb_backtester.py inside kalshi-basket-backtester using PyCharm’s file browser.
  • Add the imports and input helpers by pasting this code into arb_backtester.py:
import argparse
import csv
import json
from pathlib import Path

import matplotlib.pyplot as plt
from scipy.optimize import linprog


def load_config(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_snapshots(path):
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    return [json.loads(line) for line in lines if line.strip()]


def ordered_legs(config):
    states = config["states"]
    if not states:
        raise ValueError("At least one feasible settlement state is required.")

    legs = set(states[0]["payouts"])
    for state in states:
        if set(state["payouts"]) != legs:
            raise ValueError("Every state must contain the same payoff legs.")
    return sorted(legs)

What Do These Helpers Protect?

  • The loaders turn constraints.json and each non-empty snapshots.jsonl line into Python data.
  • The ordered_legs() function checks that every state defines exactly the same contracts.
  • Sorting the legs gives every price and payoff a stable column position.
  • Save arb_backtester.py.
  • Inspect the file in PyCharm to confirm no syntax errors are highlighted.

Seeing an Import or Syntax Error?

  • Check that every import starts at the left edge of arb_backtester.py.
  • Confirm the virtual environment from the previous step remains activated.
  • Compare the indentation inside ordered_legs() with the code block.
  • Help me debug the loaders in arb_backtester.py.
Build the payoff-constrained optimizer

The optimizer represents each state as one row of payouts. SciPy minimizes quote cost while each negative payoff row enforces a payout of at least 1.0.

  • Add the first part of solve_snapshot() below ordered_legs() in arb_backtester.py:
def solve_snapshot(config, snapshot, execution_buffer):
    legs = ordered_legs(config)
    quotes = snapshot["quotes"]
    missing = [leg for leg in legs if leg not in quotes]
    if missing:
        raise ValueError(f"Snapshot {snapshot['timestamp']} is missing: {missing}")

    prices = [float(quotes[leg]["ask"]) for leg in legs]
    depths = [float(quotes[leg]["depth"]) for leg in legs]
    payoff_matrix = [
        [float(state["payouts"][leg]) for leg in legs]
        for state in config["states"]
    ]

    result = linprog(
        prices,
        A_ub=[[-value for value in row] for row in payoff_matrix],
        b_ub=[-1.0] * len(payoff_matrix),
        bounds=[(0.0, depth) for depth in depths],
        method="highs",
    )

How Are the Constraints Built?

  • The prices list becomes the cost objective that linprog() minimizes.
  • Each row in payoff_matrix describes what every contract pays in one feasible state.
  • Negating each payoff row converts a minimum-payout rule into the upper-bound form expected by the optimizer.
  • Each bounds pair keeps a quantity non-negative and below the displayed depth.
  • Save arb_backtester.py.
  • Inspect the new function to confirm the closing parenthesis for linprog() is aligned with result.

Is the Optimizer Call Highlighted?

  • Check that linprog is imported from scipy.optimize.
  • Confirm every line inside the function uses four spaces of indentation.
  • Help me fix the linprog setup in solve_snapshot().

A solver can also report that no portfolio satisfies the constraints. The result object needs a consistent shape in that case.

  • Continue solve_snapshot() directly below the linprog() call by adding this code:
    if not result.success:
        return {
            "timestamp": snapshot["timestamp"],
            "status": "INFEASIBLE",
            "cost": "",
            "min_payout": "",
            "gross_edge": "",
            "net_edge": "",
            "capacity": "",
            "theoretical_pnl": "",
            "positions": "",
        }

    quantities = [max(0.0, float(value)) for value in result.x]
    state_payouts = [
        sum(payoff * quantity for payoff, quantity in zip(row, quantities))
        for row in payoff_matrix
    ]
    min_payout = min(state_payouts)

How Is the Guarantee Checked?

  • An infeasible result returns empty metrics so callers can handle it without reading missing values.
  • The quantities list contains the selected amount of each sorted leg.
  • The state_payouts calculation tests the selected basket against every payoff row.
  • The smallest state payout becomes the basket’s guarantee.
  • Save arb_backtester.py.
  • Inspect the return dictionary to confirm each metric key has a matching value.

Seeing an Indentation Error?

  • Place the if not result.success: line at the same indentation level as result = linprog(.
  • Indent each dictionary entry beneath return {.
  • Help me fix the result-handling indentation.

The remaining calculations turn a feasible solution into metrics that the runner can print. Capacity comes from the tightest depth-to-quantity ratio among active legs.

  • Finish solve_snapshot() by adding this code below min_payout = min(state_payouts):
    cost = float(result.fun)
    gross_edge = min_payout - cost
    net_edge = gross_edge - execution_buffer
    active = [
        (leg, quantity, depth)
        for leg, quantity, depth in zip(legs, quantities, depths)
        if quantity > 1e-8
    ]
    capacity = min(depth / quantity for _, quantity, depth in active)
    status = "OPPORTUNITY" if net_edge > 1e-9 else "REJECTED"
    theoretical_pnl = net_edge * capacity if status == "OPPORTUNITY" else 0.0
    positions = "; ".join(f"{leg}={quantity:.4f}" for leg, quantity, _ in active)

    return {
        "timestamp": snapshot["timestamp"],
        "status": status,
        "cost": cost,
        "min_payout": min_payout,
        "gross_edge": gross_edge,
        "net_edge": net_edge,
        "capacity": capacity,
        "theoretical_pnl": theoretical_pnl,
        "positions": positions,
    }

What Does the Solution Record?

  • The cost comes directly from the minimized objective value.
  • The gross edge subtracts cost from the minimum state payout.
  • The active positions retain only quantities above the numerical tolerance.
  • The returned dictionary keeps the timestamp and portfolio metrics together.
  • Save arb_backtester.py.
  • Inspect the end of solve_snapshot() to confirm the return dictionary remains inside the function.

Is the Function Ending Too Early?

  • Keep cost = float(result.fun) aligned with min_payout = min(state_payouts).
  • Keep the final return { indented inside solve_snapshot().
  • Help me debug the solution metrics in solve_snapshot().
Run the depth-bounded basket

The reusable result helpers keep each optimized row consistent across terminal output, CSV output, summary metrics, and charts. Adding them now gives the runner one result format to work with.

  • Add the CSV writer below solve_snapshot() in arb_backtester.py:
def write_results(path, results):
    fieldnames = [
        "timestamp",
        "status",
        "cost",
        "min_payout",
        "gross_edge",
        "net_edge",
        "capacity",
        "theoretical_pnl",
        "positions",
    ]
    with Path(path).open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)

What Does the Writer Preserve?

  • The field list fixes the output column order for every optimized snapshot.
  • The CSV writer stores the same dictionary keys returned by solve_snapshot().
  • Save arb_backtester.py.
  • Inspect write_results() to confirm the field names match the solution dictionary.

Is the CSV Helper Highlighted?

  • Confirm csv appears in the import group.
  • Check that writer.writeheader() remains inside the file context.
  • Help me fix write_results().
  • Add the summary helper below write_results() in arb_backtester.py:
def print_summary(results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    opportunities = [row for row in feasible if row["status"] == "OPPORTUNITY"]
    average_edge = (
        sum(row["net_edge"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    average_capacity = (
        sum(row["capacity"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    total_pnl = sum(row["theoretical_pnl"] for row in opportunities)

    print(f"Snapshots: {len(results)}")
    print(f"Opportunities: {len(opportunities)}")
    print(f"Average positive net edge: ${average_edge:.4f}")
    print(f"Average opportunity capacity: {average_capacity:.2f} normalized baskets")
    print(f"Cumulative hypothetical P&L: ${total_pnl:.4f}")

How Are Summary Values Protected?

  • Infeasible rows stay out of calculations that require numeric metrics.
  • Empty opportunity sets produce zero averages without dividing by zero.
  • The printed totals use the same fields returned by the optimizer.
  • Save arb_backtester.py.
  • Inspect the two parenthesized averages to confirm each closing parenthesis is aligned.

Seeing an Error in the Averages?

  • Keep each conditional expression inside its matching parentheses.
  • Check that opportunities uses the exact spelling from its assignment.
  • Help me debug print_summary().
  • Add the chart helper below print_summary() in arb_backtester.py:
def write_chart(path, results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    indices = list(range(len(feasible)))
    gross_edges = [row["gross_edge"] for row in feasible]
    cumulative_pnl = []
    running_total = 0.0
    for row in feasible:
        running_total += row["theoretical_pnl"]
        cumulative_pnl.append(running_total)

    plt.plot(indices, gross_edges, label="Gross edge per guaranteed dollar")
    plt.plot(indices, cumulative_pnl, label="Cumulative hypothetical P&L")
    plt.legend()
    plt.savefig(path, bbox_inches="tight")

What Does the Chart Helper Prepare?

  • The first series tracks gross edge across feasible snapshots.
  • The second series accumulates hypothetical profit across the replay.
  • The labels give the saved figure a legend.
  • Save arb_backtester.py.
  • Inspect the chart helper to confirm both plot calls use the same indices list.

Is Matplotlib Undefined?

  • Confirm matplotlib.pyplot is imported as plt.
  • Check that the activated environment contains the requirements installed earlier.
  • Help me fix write_chart().

The command-line runner loads the existing files and solves each snapshot with the same constraint model. Its first printed row provides the end-of-step guarantee check.

  • Add main() below write_chart() in arb_backtester.py:
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--constraints", default="constraints.json")
    parser.add_argument("--snapshots", default="snapshots.jsonl")
    parser.add_argument("--buffer", type=float, default=0.01)
    args = parser.parse_args()

    config = load_config(args.constraints)
    snapshots = load_snapshots(args.snapshots)
    results = [solve_snapshot(config, snapshot, args.buffer) for snapshot in snapshots]

    for row in results:
        if row["status"] == "INFEASIBLE":
            print(f"{row['timestamp']} INFEASIBLE")
            continue
        print(
            f"{row['timestamp']} {row['status']}: "
            f"cost=${row['cost']:.4f}, min payout=${row['min_payout']:.4f}, "
            f"net edge=${row['net_edge']:.4f}, capacity={row['capacity']:.2f}, "
            f"legs={row['positions']}"
        )


if __name__ == "__main__":
    main()

How Does the Runner Expose the Result?

  • The parser connects the terminal arguments to the existing constraint and snapshot files.
  • Each snapshot passes through the same solve_snapshot() function.
  • Every feasible row prints its cost, minimum payout, net edge, capacity, and selected legs.
  • Save arb_backtester.py.

Before you run the optimizer, which contracts do you think can cover all three settlement rows at the lowest cost?

  • Run the constraint-aware optimizer with the existing synthetic files by running this command:
python3 arb_backtester.py --constraints constraints.json --snapshots snapshots.jsonl

The first printed row shows a minimum payout of $1.0000. It selects fractional quantities of SYN-A:NO, SYN-B:NO, and SYN-ANY:YES.

That’s the key shift complete. Your cheapest basket now survives every settlement state represented in constraints.json.

Not Seeing a $1.0000 Minimum Payout?

  • Confirm the command runs from inside kalshi-basket-backtester.
  • Check that constraints.json still contains all three feasible states.
  • Verify snapshots.jsonl still contains the six quote legs in every snapshot.
  • Help me diagnose the optimizer output.

✔️ Awesome, I've got everything!

Great. Confirm arb_backtester.py is saved and the first optimized row maintains a $1.0000 minimum payout.

ⓧ I'd like to double check the full code

Compare your saved arb_backtester.py with this cumulative version.

import argparse
import csv
import json
from pathlib import Path

import matplotlib.pyplot as plt
from scipy.optimize import linprog


def load_config(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_snapshots(path):
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    return [json.loads(line) for line in lines if line.strip()]


def ordered_legs(config):
    states = config["states"]
    if not states:
        raise ValueError("At least one feasible settlement state is required.")

    legs = set(states[0]["payouts"])
    for state in states:
        if set(state["payouts"]) != legs:
            raise ValueError("Every state must contain the same payoff legs.")
    return sorted(legs)


def solve_snapshot(config, snapshot, execution_buffer):
    legs = ordered_legs(config)
    quotes = snapshot["quotes"]
    missing = [leg for leg in legs if leg not in quotes]
    if missing:
        raise ValueError(f"Snapshot {snapshot['timestamp']} is missing: {missing}")

    prices = [float(quotes[leg]["ask"]) for leg in legs]
    depths = [float(quotes[leg]["depth"]) for leg in legs]
    payoff_matrix = [
        [float(state["payouts"][leg]) for leg in legs]
        for state in config["states"]
    ]

    result = linprog(
        prices,
        A_ub=[[-value for value in row] for row in payoff_matrix],
        b_ub=[-1.0] * len(payoff_matrix),
        bounds=[(0.0, depth) for depth in depths],
        method="highs",
    )

    if not result.success:
        return {
            "timestamp": snapshot["timestamp"],
            "status": "INFEASIBLE",
            "cost": "",
            "min_payout": "",
            "gross_edge": "",
            "net_edge": "",
            "capacity": "",
            "theoretical_pnl": "",
            "positions": "",
        }

    quantities = [max(0.0, float(value)) for value in result.x]
    state_payouts = [
        sum(payoff * quantity for payoff, quantity in zip(row, quantities))
        for row in payoff_matrix
    ]
    min_payout = min(state_payouts)
    cost = float(result.fun)
    gross_edge = min_payout - cost
    net_edge = gross_edge - execution_buffer
    active = [
        (leg, quantity, depth)
        for leg, quantity, depth in zip(legs, quantities, depths)
        if quantity > 1e-8
    ]
    capacity = min(depth / quantity for _, quantity, depth in active)
    status = "OPPORTUNITY" if net_edge > 1e-9 else "REJECTED"
    theoretical_pnl = net_edge * capacity if status == "OPPORTUNITY" else 0.0
    positions = "; ".join(f"{leg}={quantity:.4f}" for leg, quantity, _ in active)

    return {
        "timestamp": snapshot["timestamp"],
        "status": status,
        "cost": cost,
        "min_payout": min_payout,
        "gross_edge": gross_edge,
        "net_edge": net_edge,
        "capacity": capacity,
        "theoretical_pnl": theoretical_pnl,
        "positions": positions,
    }


def write_results(path, results):
    fieldnames = [
        "timestamp",
        "status",
        "cost",
        "min_payout",
        "gross_edge",
        "net_edge",
        "capacity",
        "theoretical_pnl",
        "positions",
    ]
    with Path(path).open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)


def print_summary(results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    opportunities = [row for row in feasible if row["status"] == "OPPORTUNITY"]
    average_edge = (
        sum(row["net_edge"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    average_capacity = (
        sum(row["capacity"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    total_pnl = sum(row["theoretical_pnl"] for row in opportunities)

    print(f"Snapshots: {len(results)}")
    print(f"Opportunities: {len(opportunities)}")
    print(f"Average positive net edge: ${average_edge:.4f}")
    print(f"Average opportunity capacity: {average_capacity:.2f} normalized baskets")
    print(f"Cumulative hypothetical P&L: ${total_pnl:.4f}")


def write_chart(path, results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    indices = list(range(len(feasible)))
    gross_edges = [row["gross_edge"] for row in feasible]
    cumulative_pnl = []
    running_total = 0.0
    for row in feasible:
        running_total += row["theoretical_pnl"]
        cumulative_pnl.append(running_total)

    plt.plot(indices, gross_edges, label="Gross edge per guaranteed dollar")
    plt.plot(indices, cumulative_pnl, label="Cumulative hypothetical P&L")
    plt.legend()
    plt.savefig(path, bbox_inches="tight")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--constraints", default="constraints.json")
    parser.add_argument("--snapshots", default="snapshots.jsonl")
    parser.add_argument("--buffer", type=float, default=0.01)
    args = parser.parse_args()

    config = load_config(args.constraints)
    snapshots = load_snapshots(args.snapshots)
    results = [solve_snapshot(config, snapshot, args.buffer) for snapshot in snapshots]

    for row in results:
        if row["status"] == "INFEASIBLE":
            print(f"{row['timestamp']} INFEASIBLE")
            continue
        print(
            f"{row['timestamp']} {row['status']}: "
            f"cost=${row['cost']:.4f}, min payout=${row['min_payout']:.4f}, "
            f"net edge=${row['net_edge']:.4f}, capacity={row['capacity']:.2f}, "
            f"legs={row['positions']}"
        )


if __name__ == "__main__":
    main()

Your optimizer now converts settlement rules into enforceable payout constraints. Next, you’ll replay the quote snapshots to test whether the signal survives across time and available liquidity.

Replay Snapshots as a Backtest

Your linear programming solver now finds a depth-bounded basket that pays at least $1 in every feasible settlement state. That proof comes from one quote snapshot.

Replay backtesting sends every timestamped JSONL quote set through the same solver. This reveals whether the edge persists over time.

The project-defined execution haircut reduces each gross edge by $0.01 per guaranteed dollar. Depth-based capacity limits the hypothetical result to the quoted liquidity.

In this step, get ready to:
  • Replay all four quote snapshots with a configurable execution haircut.
  • Compare depth-based capacity across accepted and rejected decisions.
  • Export every decision to a CSV result set.
Trace the replay implementation

The existing solve_snapshot() function turns one quote snapshot into one decision row. The replay path applies that calculation to every entry in snapshots.jsonl.

  • Switch back to arb_backtester.py in the PyCharm window from earlier.
  • Compare the replay implementation with the full-file reference below.

✔️ Awesome, I've got everything!

Your arb_backtester.py already contains the replay loop, execution haircut, capacity calculation, decision classification, and CSV writer.

ⓧ I'd like to double check the full code

Compare your saved arb_backtester.py with this complete reference.

import argparse
import csv
import json
from pathlib import Path

import matplotlib.pyplot as plt
from scipy.optimize import linprog


def load_config(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_snapshots(path):
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    return [json.loads(line) for line in lines if line.strip()]


def ordered_legs(config):
    states = config["states"]
    if not states:
        raise ValueError("At least one feasible settlement state is required.")

    legs = set(states[0]["payouts"])
    for state in states:
        if set(state["payouts"]) != legs:
            raise ValueError("Every state must contain the same payoff legs.")
    return sorted(legs)


def solve_snapshot(config, snapshot, execution_buffer):
    legs = ordered_legs(config)
    quotes = snapshot["quotes"]
    missing = [leg for leg in legs if leg not in quotes]
    if missing:
        raise ValueError(f"Snapshot {snapshot['timestamp']} is missing: {missing}")

    prices = [float(quotes[leg]["ask"]) for leg in legs]
    depths = [float(quotes[leg]["depth"]) for leg in legs]
    payoff_matrix = [
        [float(state["payouts"][leg]) for leg in legs]
        for state in config["states"]
    ]

    result = linprog(
        prices,
        A_ub=[[-value for value in row] for row in payoff_matrix],
        b_ub=[-1.0] * len(payoff_matrix),
        bounds=[(0.0, depth) for depth in depths],
        method="highs",
    )

    if not result.success:
        return {
            "timestamp": snapshot["timestamp"],
            "status": "INFEASIBLE",
            "cost": "",
            "min_payout": "",
            "gross_edge": "",
            "net_edge": "",
            "capacity": "",
            "theoretical_pnl": "",
            "positions": "",
        }

    quantities = [max(0.0, float(value)) for value in result.x]
    state_payouts = [
        sum(payoff * quantity for payoff, quantity in zip(row, quantities))
        for row in payoff_matrix
    ]
    min_payout = min(state_payouts)
    cost = float(result.fun)
    gross_edge = min_payout - cost
    net_edge = gross_edge - execution_buffer
    active = [
        (leg, quantity, depth)
        for leg, quantity, depth in zip(legs, quantities, depths)
        if quantity > 1e-8
    ]
    capacity = min(depth / quantity for _, quantity, depth in active)
    status = "OPPORTUNITY" if net_edge > 1e-9 else "REJECTED"
    theoretical_pnl = net_edge * capacity if status == "OPPORTUNITY" else 0.0
    positions = "; ".join(f"{leg}={quantity:.4f}" for leg, quantity, _ in active)

    return {
        "timestamp": snapshot["timestamp"],
        "status": status,
        "cost": cost,
        "min_payout": min_payout,
        "gross_edge": gross_edge,
        "net_edge": net_edge,
        "capacity": capacity,
        "theoretical_pnl": theoretical_pnl,
        "positions": positions,
    }


def write_results(path, results):
    fieldnames = [
        "timestamp",
        "status",
        "cost",
        "min_payout",
        "gross_edge",
        "net_edge",
        "capacity",
        "theoretical_pnl",
        "positions",
    ]
    with Path(path).open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)


def print_summary(results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    opportunities = [row for row in feasible if row["status"] == "OPPORTUNITY"]
    average_edge = (
        sum(row["net_edge"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    average_capacity = (
        sum(row["capacity"] for row in opportunities) / len(opportunities)
        if opportunities
        else 0.0
    )
    total_pnl = sum(row["theoretical_pnl"] for row in opportunities)

    print(f"Snapshots: {len(results)}")
    print(f"Opportunities: {len(opportunities)}")
    print(f"Average positive net edge: ${average_edge:.4f}")
    print(f"Average opportunity capacity: {average_capacity:.2f} normalized baskets")
    print(f"Cumulative hypothetical P&L: ${total_pnl:.4f}")


def write_chart(path, results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    indices = list(range(len(feasible)))
    gross_edges = [row["gross_edge"] for row in feasible]
    cumulative_pnl = []
    running_total = 0.0
    for row in feasible:
        running_total += row["theoretical_pnl"]
        cumulative_pnl.append(running_total)

    plt.plot(indices, gross_edges, label="Gross edge per guaranteed dollar")
    plt.plot(indices, cumulative_pnl, label="Cumulative hypothetical P&L")
    plt.legend()
    plt.savefig(path, bbox_inches="tight")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--constraints", default="constraints.json")
    parser.add_argument("--snapshots", default="snapshots.jsonl")
    parser.add_argument("--buffer", type=float, default=0.01)
    parser.add_argument("--output", default="backtest_results.csv")
    parser.add_argument("--chart", default="arb_report.png")
    args = parser.parse_args()

    config = load_config(args.constraints)
    snapshots = load_snapshots(args.snapshots)
    results = [solve_snapshot(config, snapshot, args.buffer) for snapshot in snapshots]

    for row in results:
        if row["status"] == "INFEASIBLE":
            print(f"{row['timestamp']} INFEASIBLE")
            continue
        print(
            f"{row['timestamp']} {row['status']}: "
            f"cost=${row['cost']:.4f}, min payout=${row['min_payout']:.4f}, "
            f"net edge=${row['net_edge']:.4f}, capacity={row['capacity']:.2f}, "
            f"legs={row['positions']}"
        )

    write_results(args.output, results)
    print_summary(results)
    write_chart(args.chart, results)
    print(f"Wrote {args.output} and {args.chart}")


if __name__ == "__main__":
    main()

What should this implementation preserve?

  • The INFEASIBLE branch preserves the timestamp while leaving unavailable metrics empty.
  • The net_edge calculation subtracts execution_buffer from gross_edge.
  • The capacity calculation finds the smallest active-leg depth-to-quantity ratio.
  • The theoretical_pnl calculation produces a positive result only for an OPPORTUNITY row.
  • The main() function solves every snapshot before passing the complete result list to write_results().
Run the four-snapshot replay

Each invocation uses the same settlement constraints against a new synchronized quote set. A snapshot becomes an OPPORTUNITY only when its edge remains positive after the execution haircut.

Before you run the replay, predict whether every positive gross edge will survive the 0.01 execution haircut.

  • Switch back to the activated terminal from earlier.
  • Replay the four synthetic snapshots by running this command:
python3 arb_backtester.py --constraints constraints.json --snapshots snapshots.jsonl --buffer 0.01 --output backtest_results.csv

What does this replay do?

The --buffer 0.01 argument subtracts the project-defined haircut from each normalized basket's gross edge. This sensitivity allowance does not represent an actual fee or fill guarantee.

The --output backtest_results.csv argument sends all four decision rows to one audit file. Each row preserves the quote timestamp that produced it.

You should see four timestamped decision lines. The lines include both OPPORTUNITY and REJECTED decisions.

That is the replay working. One pricing print no longer defines the whole strategy.

Replay command not working?

  • Confirm the terminal prompt still shows the activated .venv environment.
  • Confirm constraints.json contains the same six payoff legs found in every quote object inside snapshots.jsonl.
  • Confirm the terminal is still inside the kalshi-basket-backtester folder.
  • Help me diagnose my four-snapshot replay.
Inspect the CSV decision history

The write_results() function turns each solver result into an auditable record. Its separate edge and capacity fields keep a theoretical signal distinct from its depth-limited scale.

Before you inspect the file, predict which fields reveal whether a positive basket remains usable after the execution haircut.

  • Return to the PyCharm window from earlier.
  • Select backtest_results.csv from the project file list.
  • Confirm the header reads timestamp,status,cost,min_payout,gross_edge,net_edge,capacity,theoretical_pnl,positions.
  • Count the four timestamped data rows beneath the header.
  • Find an OPPORTUNITY value in the status column.
  • Find a REJECTED value in the status column.

You should see nine columns across four timestamped rows. The status column contains both decision types.

How do capacity and profit connect?

Each solver portfolio guarantees one normalized payout unit. The active-leg depth-to-quantity ratios reveal how many copies of that portfolio fit inside the displayed depth.

The smallest ratio becomes capacity. An accepted row multiplies that value by net_edge to estimate theoretical_pnl.

CSV missing or incomplete?

  • Check the terminal output for a final line confirming the written output paths.
  • Refresh PyCharm's project file list if backtest_results.csv is not visible.
  • Confirm snapshots.jsonl still contains four non-empty lines.
  • Help me debug my missing backtest CSV.

Your replay now separates recurring opportunities from rejected snapshots. Next up, you'll turn those rows into summary metrics and a chart.

Turn Results into Analytics

Your replay engine now classifies all four snapshots in backtest_results.csv. Each row records the edge that remains after the execution haircut.

A row-by-row CSV can hide whether an edge recurs. Summary metrics plus a chart reveal whether the signal has useful capacity across time.

You will use Matplotlib to turn the replay into two visible trends. This final reporting layer separates recurring opportunities from a single attractive print.

In this step, get ready to:
  • Calculate opportunity frequency plus portfolio performance metrics.
  • Plot gross edge beside cumulative hypothetical profit.
  • Compare the rejected snapshot with the profitable snapshots.
Calculate the portfolio summary

A useful summary measures how often the optimizer finds an opportunity. It also measures the size of the opportunities that survive the execution haircut.

  • In PyCharm, select arb_backtester.py from the project file tree.
  • Locate the end of write_results().
  • Add the summary function below write_results() by pasting this code:
def print_summary(results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    opportunities = [row for row in feasible if row["status"] == "OPPORTUNITY"]
    average_edge = sum(row["net_edge"] for row in opportunities) / len(opportunities) if opportunities else 0.0
    average_capacity = sum(row["capacity"] for row in opportunities) / len(opportunities) if opportunities else 0.0
    total_pnl = sum(row["theoretical_pnl"] for row in opportunities)
    print(f"Snapshots: {len(results)}")
    print(f"Opportunities: {len(opportunities)}")
    print(f"Average positive net edge: ${average_edge:.4f}")
    print(f"Average opportunity capacity: {average_capacity:.2f} normalized baskets")
    print(f"Cumulative hypothetical P&L: ${total_pnl:.4f}")

What does this function calculate?

  • The feasible list excludes rows where the optimizer could not construct a guaranteed basket.
  • The opportunities list keeps rows with positive net edge after the execution haircut.
  • The two averages report surviving edge plus depth-based capacity.
  • The total_pnl value adds the hypothetical profit from every opportunity.
  • Locate write_results(args.output, results) inside main().
  • Add the summary call directly below that line using this code:
write_results(args.output, results)
print_summary(results)

How does the summary join the replay?

The replay writes every decision to the requested output file first. The print_summary() call then analyzes the same result rows.

  • Save arb_backtester.py.
  • Run the summary-enabled replay from the activated virtual environment with this command:
python3 arb_backtester.py --constraints constraints.json --snapshots snapshots.jsonl --buffer 0.01 --output backtest_results.csv

What does this run prove?

You should see Snapshots: 4 plus Opportunities: 3 below the individual decisions. You should also see the average edge plus the average capacity.

The cumulative hypothetical profit should be $5.4200. Your four replay decisions now have a portfolio-level summary.

Missing the summary lines?

  • Confirm print_summary(results) appears below write_results(args.output, results) inside main().
  • Check that print_summary() starts at the left edge of the file.
  • Make sure the virtual environment from the earlier setup remains active.
  • Help me debug missing summary output from my backtester.
Plot edge and cumulative profit

A chart exposes the timing of each signal. The gross-edge line shows the raw pricing pattern.

The cumulative-profit line grows only when a snapshot remains profitable after the haircut. A flat section makes a rejected decision visible immediately.

  • Locate print_summary() in arb_backtester.py.
  • Add the chart function below print_summary() by pasting this code:
def write_chart(path, results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    indices = list(range(len(feasible)))
    gross_edges = [row["gross_edge"] for row in feasible]
    cumulative_pnl = []
    running_total = 0.0
    for row in feasible:
        running_total += row["theoretical_pnl"]
        cumulative_pnl.append(running_total)
    plt.plot(indices, gross_edges, label="Gross edge per guaranteed dollar")
    plt.plot(indices, cumulative_pnl, label="Cumulative hypothetical P&L")
    plt.legend()
    plt.savefig(path, bbox_inches="tight")

What does the chart show?

  • The gross_edges list preserves the pricing edge from each feasible replay row.
  • The running_total value grows when an opportunity contributes positive hypothetical profit.
  • Each plt.plot() call gives its line a label for the legend.
  • The plt.savefig() call writes the current figure to the requested path.
  • Return to the main() function.
  • Replace the full main() function with this chart-enabled version:
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--constraints", default="constraints.json")
    parser.add_argument("--snapshots", default="snapshots.jsonl")
    parser.add_argument("--buffer", type=float, default=0.01)
    parser.add_argument("--output", default="backtest_results.csv")
    parser.add_argument("--chart", default="arb_report.png")
    args = parser.parse_args()
    config = load_config(args.constraints)
    snapshots = load_snapshots(args.snapshots)
    results = [solve_snapshot(config, snapshot, args.buffer) for snapshot in snapshots]
    for row in results:
        if row["status"] == "INFEASIBLE":
            print(f"{row['timestamp']} INFEASIBLE")
            continue
        print(f"{row['timestamp']} {row['status']}: cost=${row['cost']:.4f}, min payout=${row['min_payout']:.4f}, net edge=${row['net_edge']:.4f}, capacity={row['capacity']:.2f}, legs={row['positions']}")
    write_results(args.output, results)
    print_summary(results)
    write_chart(args.chart, results)
    print(f"Wrote {args.output} and {args.chart}")

How does the chart path work?

The --chart argument controls the PNG output path. The replay passes every result to write_chart() after printing the summary.

The final message confirms both output paths. Each run now leaves an auditable data file plus a visual report.

✔️ Awesome, I've got everything!

Strong progress. Your backtester now calculates summary metrics plus a chart from the same replay results.

ⓧ I'd like to double check the full code

Compare your completed arb_backtester.py with this cumulative reference.

import argparse
import csv
import json
from pathlib import Path

import matplotlib.pyplot as plt
from scipy.optimize import linprog


def load_config(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_snapshots(path):
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    return [json.loads(line) for line in lines if line.strip()]


def ordered_legs(config):
    states = config["states"]
    if not states:
        raise ValueError("At least one feasible settlement state is required.")

    legs = set(states[0]["payouts"])
    for state in states:
        if set(state["payouts"]) != legs:
            raise ValueError("Every state must contain the same payoff legs.")
    return sorted(legs)


def solve_snapshot(config, snapshot, execution_buffer):
    legs = ordered_legs(config)
    quotes = snapshot["quotes"]
    missing = [leg for leg in legs if leg not in quotes]
    if missing:
        raise ValueError(f"Snapshot {snapshot['timestamp']} is missing: {missing}")

    prices = [float(quotes[leg]["ask"]) for leg in legs]
    depths = [float(quotes[leg]["depth"]) for leg in legs]
    payoff_matrix = [[float(state["payouts"][leg]) for leg in legs] for state in config["states"]]
    result = linprog(prices, A_ub=[[-value for value in row] for row in payoff_matrix], b_ub=[-1.0] * len(payoff_matrix), bounds=[(0.0, depth) for depth in depths], method="highs")
    if not result.success:
        return {"timestamp": snapshot["timestamp"], "status": "INFEASIBLE", "cost": "", "min_payout": "", "gross_edge": "", "net_edge": "", "capacity": "", "theoretical_pnl": "", "positions": ""}
    quantities = [max(0.0, float(value)) for value in result.x]
    state_payouts = [sum(payoff * quantity for payoff, quantity in zip(row, quantities)) for row in payoff_matrix]
    min_payout = min(state_payouts)
    cost = float(result.fun)
    gross_edge = min_payout - cost
    net_edge = gross_edge - execution_buffer
    active = [(leg, quantity, depth) for leg, quantity, depth in zip(legs, quantities, depths) if quantity > 1e-8]
    capacity = min(depth / quantity for _, quantity, depth in active)
    status = "OPPORTUNITY" if net_edge > 1e-9 else "REJECTED"
    theoretical_pnl = net_edge * capacity if status == "OPPORTUNITY" else 0.0
    positions = "; ".join(f"{leg}={quantity:.4f}" for leg, quantity, _ in active)
    return {"timestamp": snapshot["timestamp"], "status": status, "cost": cost, "min_payout": min_payout, "gross_edge": gross_edge, "net_edge": net_edge, "capacity": capacity, "theoretical_pnl": theoretical_pnl, "positions": positions}


def write_results(path, results):
    fieldnames = ["timestamp", "status", "cost", "min_payout", "gross_edge", "net_edge", "capacity", "theoretical_pnl", "positions"]
    with Path(path).open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)


def print_summary(results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    opportunities = [row for row in feasible if row["status"] == "OPPORTUNITY"]
    average_edge = sum(row["net_edge"] for row in opportunities) / len(opportunities) if opportunities else 0.0
    average_capacity = sum(row["capacity"] for row in opportunities) / len(opportunities) if opportunities else 0.0
    total_pnl = sum(row["theoretical_pnl"] for row in opportunities)
    print(f"Snapshots: {len(results)}")
    print(f"Opportunities: {len(opportunities)}")
    print(f"Average positive net edge: ${average_edge:.4f}")
    print(f"Average opportunity capacity: {average_capacity:.2f} normalized baskets")
    print(f"Cumulative hypothetical P&L: ${total_pnl:.4f}")


def write_chart(path, results):
    feasible = [row for row in results if row["status"] != "INFEASIBLE"]
    indices = list(range(len(feasible)))
    gross_edges = [row["gross_edge"] for row in feasible]
    cumulative_pnl = []
    running_total = 0.0
    for row in feasible:
        running_total += row["theoretical_pnl"]
        cumulative_pnl.append(running_total)
    plt.plot(indices, gross_edges, label="Gross edge per guaranteed dollar")
    plt.plot(indices, cumulative_pnl, label="Cumulative hypothetical P&L")
    plt.legend()
    plt.savefig(path, bbox_inches="tight")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--constraints", default="constraints.json")
    parser.add_argument("--snapshots", default="snapshots.jsonl")
    parser.add_argument("--buffer", type=float, default=0.01)
    parser.add_argument("--output", default="backtest_results.csv")
    parser.add_argument("--chart", default="arb_report.png")
    args = parser.parse_args()
    config = load_config(args.constraints)
    snapshots = load_snapshots(args.snapshots)
    results = [solve_snapshot(config, snapshot, args.buffer) for snapshot in snapshots]
    for row in results:
        if row["status"] == "INFEASIBLE":
            print(f"{row['timestamp']} INFEASIBLE")
            continue
        print(f"{row['timestamp']} {row['status']}: cost=${row['cost']:.4f}, min payout=${row['min_payout']:.4f}, net edge=${row['net_edge']:.4f}, capacity={row['capacity']:.2f}, legs={row['positions']}")
    write_results(args.output, results)
    print_summary(results)
    write_chart(args.chart, results)
    print(f"Wrote {args.output} and {args.chart}")


if __name__ == "__main__":
    main()

How to use this reference

Check the order of the helper functions plus the calls inside main(). The reference matches the completed analytics engine.

Run and compare the analytics

The completed run reveals whether attractive prices persist across time. Before the run, think about whether cumulative hypothetical profit will rise at the rejected snapshot.

  • Save arb_backtester.py.
  • Generate the refreshed CSV plus the analytics chart by running this command:
python3 arb_backtester.py --constraints constraints.json --snapshots snapshots.jsonl --buffer 0.01 --output backtest_results.csv --chart arb_report.png

What should the terminal report?

  • You should see Snapshots: 4.
  • You should see Opportunities: 3.
  • You should see an average positive net edge of $0.1033.
  • You should see an average opportunity capacity of 15.33 normalized baskets.
  • You should see cumulative hypothetical profit of $5.4200.
  • The final line should confirm that the replay wrote backtest_results.csv plus arb_report.png.
  • Double-click arb_report.png in the PyCharm file tree.

You should see a gross-edge line plus a cumulative hypothetical-profit line. The gross edge falls below zero at the second snapshot.

The cumulative-profit line stays flat at that point. The rejected row contributes no hypothetical profit.

  • Select backtest_results.csv in the PyCharm file tree.
  • Locate the row with timestamp 2026-09-18T10:01:00Z.

Why was this snapshot rejected?

The optimized basket uses SYN-A:NO at 0.60. It also uses SYN-B:NO at 0.59.

The third active leg is SYN-ANY:YES at 0.83. The normalized basket costs $1.0100.

Its minimum payout remains $1.0000. The $0.0100 execution haircut lowers net edge to -$0.0200.

The state guarantee still holds. The price combination removes the profitable edge after the execution allowance.

Chart missing or blank?

  • Confirm import matplotlib.pyplot as plt appears above the SciPy import.
  • Check that write_chart(args.chart, results) appears after print_summary(results) inside main().
  • Verify that write_chart() calls plt.savefig(path, bbox_inches="tight").
  • Help me debug a missing or blank arb_report.png chart.

That completes the research engine. You can now trace every synthetic decision from feasible-state constraints to the final chart.

Secret mission

Capture a Live Demo Basket

Your synthetic engine proves that settlement constraints can expose or reject a basket. This mission connects that engine to authenticated Kalshi Demo quotes, reconstructs top-of-book asks, and replays the result without placing an order.

Clean Up Your Resources

Clean Up Your Resources

Everything from this project lives on your Mac. There are no ongoing costs, cloud resources, or real-money orders.

You can keep the workspace, pause the active environment, or delete the local project folder.

Resources you used:

  • The kalshi-basket-backtester folder on your Desktop. This folder contains the .venv virtual environment. It also contains your source files, synthetic snapshots, demo snapshots, CSV reports, and PNG charts.

Keep everything running

No action is needed. Choose this option if you plan to replay more snapshots or continue developing the research engine.

  • Keep the kalshi-basket-backtester folder on your Desktop.
  • Leave your Kalshi Demo credential active only if you plan to capture more research snapshots.
  • Continue treating every demo result as a sequential research observation. It does not represent an order or fill.

Pause - I'll come back to this later

Pausing exits the active Python environment while preserving every project file. No backtester or collector continues running in the background.

  • Deactivate the current virtual environment by running:
deactivate

What Does This Command Do?

This exits the current virtual environment. The .venv directory remains inside your project folder for your next session.

  • Keep the kalshi-basket-backtester folder on your Desktop so you can resume from the same files later.

Delete - I don't want to use this again

Deleting the folder is permanent. Your Kalshi Demo account and any RSA private key stored elsewhere remain untouched.

  • Exit the active virtual environment by running:
deactivate

What Does This Command Do?

This returns Terminal to your normal Python environment. It leaves the project files in place for the deletion command.

  • Move to your Desktop and delete the entire project folder by running:
cd ~/Desktop
rm -rf kalshi-basket-backtester

What Does This Cleanup Do?

The first line moves Terminal outside the project folder. The second line removes the workspace with its virtual environment, code, snapshots, reports, and charts.

  • Confirm that the folder has been removed by running:
ls ~/Desktop

What Does This Check Do?

This lists the folders on your Desktop so you can verify that the project workspace has disappeared.

You should no longer see kalshi-basket-backtester in the output. That clears every local artifact created inside the workspace.

The project used authenticated read requests without placing orders. You can also revoke the demo credential if you no longer need it.

  • Sign in to Kalshi Demo.
  • Open the settings area for your demo credentials.
  • Locate the credential identified by the API key ID used for this project.
  • Use the credential's revocation control.
  • Confirm that the credential no longer appears as active.

Nice Work!

Nice Work!

You did it! You built a constraint-aware basket arbitrage backtester that tests every feasible settlement state before accepting an opportunity.

You've learned how to:

  • Build an explicit payoff matrix that includes the NONE state. You exposed the false $0.0700 headline edge when that state paid $0.0000.
  • Use SciPy linear programming to minimize basket cost under hard payout constraints. You capped each position with quoted depth.
  • Replay JSONL snapshots with a configurable execution haircut. You exported auditable decisions to backtest_results.csv. You plotted gross edge plus cumulative hypothetical profit in arb_report.png.
  • Complete the optional Secret Mission by authenticating read-only requests to Kalshi Demo. You reconstructed YES asks from NO bids. You reconstructed NO asks from YES bids. You produced live_results.csv plus live_report.png without placing orders.

Ready to quiz yourself?