AI-Only Credit Spread¶
ai_credit_spread.py is a minimal AI-only options strategy. Python creates
one agent and runs it each iteration. The prompt defines only the credit-spread
policy. LumiBot’s built-in options skill provides reusable contract, pricing,
atomic-order, signed-position, and close-verification mechanics.
How it works¶
The agent selects a listed put or call vertical from current evidence.
It verifies both exact contracts and calculates the per-unit package credit.
It submits both legs atomically and verifies the exact order and positions.
Active rules prevent duplicate structures and repeated closing orders.
Verified backtest evidence¶
The preserved pre-fix run demonstrated the original failure clearly: reversed closing sides, repeated close attempts, and quantities that escalated to 480 contracts. That artifact is retained as the red baseline.
The repaired real-model eval now passes three consecutive repetitions. Each run
reconstructs the signed spread, maps long legs to sell_to_close and short
legs to buy_to_close, submits one correctly sized atomic close, and verifies
the final state. The current historical downloader returned no option chain for
the canonical local window, so that backtest correctly remained flat. These
results validate mechanics without claiming strategy profitability.
export GEMINI_API_KEY="your-key"
export BACKTESTING_DATA_SOURCE="ThetaData"
export DATADOWNLOADER_BASE_URL="https://<your-downloader-host>:8080"
export DATADOWNLOADER_API_KEY="your-downloader-key"
python -m lumibot.example_strategies.ai_credit_spread
Set BACKTESTING_START and BACKTESTING_END to choose an exact window.
1"""AI-only vertical credit-spread strategy driven by a LumiBot agent."""
2
3import os
4from datetime import datetime, timedelta
5from pathlib import Path
6
7from lumibot.strategies.strategy import Strategy
8
9
10def build_credit_spread_system_prompt(params: dict) -> str:
11 underlying = str(params.get("underlying", "SPY")).upper()
12 return f"""
13You are the complete decision-maker for an AI-only {underlying} vertical credit
14spread strategy. Use the LumiBot options skill for mechanics and execution.
15
16Strategy policy:
17- Prefer a {params['preferred_side']} credit spread. Switch sides only when
18 current evidence clearly supports it.
19- Prefer {params['preferred_dte']} DTE and require {params['min_dte']} to
20 {params['max_dte']} DTE.
21- Select the short leg near {params['target_delta']} absolute delta, verified
22 within {params['delta_band']} of the target.
23- Use a listed long wing exactly {params['wing_width']} points farther OTM.
24- Require a net credit between zero and the wing width.
25- Risk no more than {params['max_risk_pct']:.2%} of portfolio value and never
26 exceed {params['max_contracts']} contracts.
27- Hold at most one {underlying} option structure and manage it before new entries.
28- Close when {params['profit_take_fraction']:.0%} of credit is captured, closing
29 debit reaches {params['loss_multiple']} times opening credit, DTE is
30 {params['time_stop_dte']} or less, or short absolute delta reaches 0.30.
31- Use a no-trade decision whenever current evidence cannot prove every condition.
32
33You own research, contract selection, sizing, atomic order construction,
34submission, verification, and management. Python contains no trading decisions.
35""".strip()
36
37
38class AICreditSpreadStrategy(Strategy):
39 parameters = {
40 "underlying": "SPY", "preferred_side": "put", "wing_width": 5.0,
41 "target_delta": 0.16, "delta_band": 0.04, "min_dte": 30,
42 "max_dte": 45, "preferred_dte": 35, "profit_take_fraction": 0.50,
43 "loss_multiple": 2.0, "time_stop_dte": 21, "max_risk_pct": 0.02,
44 "max_contracts": 10,
45 }
46
47 def initialize(self):
48 self.sleeptime = "1D"
49 self.agents.create(name="credit_spread", model="gemini-3.5-flash-lite", allow_trading=True,
50 system_prompt=build_credit_spread_system_prompt(self.parameters),
51 rules_path=Path(__file__).with_name("agent_rules") / "ai_credit_spread.rules.json")
52
53 def on_trading_iteration(self):
54 self.agents["credit_spread"].run(task_prompt="Run the complete credit-spread workflow for this iteration.",
55 context={"current_datetime": self.get_datetime().isoformat(), "strategy_parameters": dict(self.parameters)})
56
57
58def _parameters_from_env(defaults: dict) -> dict:
59 params = dict(defaults)
60 if os.environ.get("AI_CS_UNDERLYING"): params["underlying"] = os.environ["AI_CS_UNDERLYING"].strip().upper()
61 if os.environ.get("AI_CS_PREFERRED_SIDE"): params["preferred_side"] = os.environ["AI_CS_PREFERRED_SIDE"].strip().lower()
62 for key in ("wing_width", "target_delta", "delta_band", "profit_take_fraction", "loss_multiple", "max_risk_pct"):
63 if os.environ.get(f"AI_CS_{key.upper()}"): params[key] = float(os.environ[f"AI_CS_{key.upper()}"])
64 for key in ("min_dte", "max_dte", "preferred_dte", "time_stop_dte", "max_contracts"):
65 if os.environ.get(f"AI_CS_{key.upper()}"): params[key] = int(os.environ[f"AI_CS_{key.upper()}"])
66 return params
67
68
69if __name__ == "__main__":
70 backtesting_end = datetime.fromisoformat(os.environ.get("BACKTESTING_END", datetime.now().date().isoformat()))
71 backtesting_start = datetime.fromisoformat(os.environ.get("BACKTESTING_START", (backtesting_end - timedelta(days=45)).date().isoformat()))
72 AICreditSpreadStrategy.backtest(None, backtesting_start=backtesting_start, backtesting_end=backtesting_end, benchmark_asset="SPY", budget=100_000, parameters=_parameters_from_env(AICreditSpreadStrategy.parameters))