"""Research signals for weather contracts. Your model, your decisions.
WHAT THIS DOES
Reads a list of places and, for each place, the contract threshold you care about
(for example "daily high above 90 F"). Pulls two things from the GetFixn / TrueFixR API:
1. live conditions at that place (current temperature, whatever else the API returns)
2. the county storm-risk grade for the next 24 hours (optional, costs more)
Then writes one row per place and contract to signals.csv and signals.json, so YOUR own
code or spreadsheet can use it.
WHAT THIS DOES NOT DO
No order placement. No calls to Kalshi or any exchange. No exchange keys, no wallet.
It only reads from data.truefixr.com and writes two local files.
The API does not return a daily-high number. It returns current conditions, an hourly temperature
forecast (add hours_ahead, 0 to 48, free in the lean response) and a county storm-risk grade. The "running high/low" column is what THIS script has seen across its own
runs today, so it is only as good as how often you run it. It is not an official reading.
READ THE CONTRACT
Every market defines its own settlement source (often an official climate report for a
named station), its own time window and its own rounding. Read those rules for each contract.
Coordinates here are a point you chose, not necessarily the settlement station. Data can lag
or differ from the settlement source. That mismatch is your risk.
GetFixn sells data, not trading advice. Nothing here is investment advice. Forecasts can be wrong.
You are responsible for your own strategies and for following the rules of any exchange or market
you use. No claim is made that this data is accurate enough to trade on, or that it helps.
Kalshi is a trademark of its owner. We have no partnership with or endorsement from any exchange.
SETUP
pip install requests
export GETFIXN_KEY=tfx_live_... (never put the key in code)
python kalshi_weather_signals.py (reads kalshi_locations.json, needs gf_client.py beside it)
COST per run: places x $0.0002 for live conditions, plus places x $0.05 if GRADES is on.
Environment knobs: LOCATIONS (file), OUT_STEM (default "signals"), STATE (default
kalshi_state.json), GRADES=0 to skip grades, BUDGET_USD (default 1.00, refuse to run above it).
"""
import csv
import json
import os
import sys
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
import gf_client as gf
PRICE_WEATHER = 0.0002 # USD per place, live conditions
PRICE_GRADE = 0.05 # USD per place, county forecast grade
OPS = ("above", "below")
FIELDS = ["ts_utc", "label", "lat", "lon", "local_date", "metric", "op", "threshold",
"value_now", "margin_now", "met_now", "extreme_so_far", "extreme_margin",
"observed_at", "age_minutes", "grade", "grade_score", "alert_count",
"settlement_source", "note", "current", "alerts"]
def load(path):
"""Read and check the locations file. Fails early with a message you can act on."""
with open(path) as f:
locs = json.load(f)
seen = set()
for l in locs:
if l.get("label") in seen or not l.get("label"):
raise ValueError("each place needs a unique label: %r" % l.get("label"))
seen.add(l["label"])
if not (-90 <= l["lat"] <= 90 and -180 <= l["lon"] <= 180):
raise ValueError("%s: lat/lon out of range" % l["label"])
ZoneInfo(l.get("tz", "UTC")) # raises if the zone name is wrong
if not l.get("contracts"):
raise ValueError("%s: add at least one contract" % l["label"])
for c in l["contracts"]:
if c.get("op") not in OPS or not isinstance(c.get("threshold"), (int, float)):
raise ValueError("%s: contract needs op above|below and a numeric threshold" % l["label"])
c.setdefault("metric", "temperature_f")
return locs
def estimate_cost(locs, grades_on):
return len(locs) * (PRICE_WEATHER + (PRICE_GRADE if grades_on else 0))
def _num(x):
return x if isinstance(x, (int, float)) and not isinstance(x, bool) else None
def _age_minutes(cur, now):
"""Minutes since the observation, if the API gave a parseable time. Otherwise None."""
for k in ("observed_at", "observation_time", "timestamp", "time"):
if cur.get(k):
try:
t = datetime.fromisoformat(str(cur[k]).replace("Z", "+00:00"))
if t.tzinfo is None:
t = t.replace(tzinfo=timezone.utc)
return str(cur[k]), round((now - t).total_seconds() / 60, 1)
except ValueError:
return str(cur[k]), None
return None, None
def _track(state, label, metric, value, local_date):
"""Running high and low for this place, this local calendar day, from our own runs only."""
s = state.setdefault(label, {}).get(metric)
if not s or s["date"] != local_date: # new day (in the place's own time zone): start over
s = {"date": local_date, "max": value, "min": value}
elif value is not None:
s["max"] = value if s["max"] is None else max(s["max"], value)
s["min"] = value if s["min"] is None else min(s["min"], value)
state[label][metric] = s
return s
def build(locs, state, now=None, window="24h", grades_on=True):
"""One API call for conditions (and one for grades). Returns rows, updates state in place."""
now = now or datetime.now(timezone.utc)
w = {r["label"]: r for r in gf.weather(locs)}
g = {r["label"]: r for r in gf.grades(locs, window)} if grades_on else {}
rows = []
for l in locs:
cur = w[l["label"]].get("current") or {}
alerts = w[l["label"]].get("alerts") or []
grade = g.get(l["label"], {}).get("grade")
local_date = now.astimezone(ZoneInfo(l.get("tz", "UTC"))).strftime("%Y-%m-%d")
obs_at, age = _age_minutes(cur, now)
for c in l["contracts"]:
v = _num(cur.get(c["metric"])) # None if the API has no such field. We never guess.
s = _track(state, l["label"], c["metric"], v, local_date)
up = c["op"] == "above"
ext = s["max"] if up else s["min"]
sign = 1 if up else -1
rows.append({
"ts_utc": now.strftime("%Y-%m-%dT%H:%M:%SZ"), "label": l["label"],
"lat": l["lat"], "lon": l["lon"], "local_date": local_date,
"metric": c["metric"], "op": c["op"], "threshold": c["threshold"],
"value_now": v,
# margin > 0 means the reading is on the contract's side of the threshold right now
"margin_now": None if v is None else round(sign * (v - c["threshold"]), 2),
"met_now": None if v is None else sign * (v - c["threshold"]) > 0,
"extreme_so_far": ext,
"extreme_margin": None if ext is None else round(sign * (ext - c["threshold"]), 2),
"observed_at": obs_at, "age_minutes": age,
"grade": grade, "grade_score": gf.GRADES.index(grade) if grade and grade in gf.GRADES else None,
"alert_count": len(alerts),
"settlement_source": c.get("settlement_source") or l.get("settlement_source"),
"note": c.get("note"),
"current": json.dumps(cur), "alerts": json.dumps(alerts)})
return rows
def write(rows, stem="signals"):
"""signals.json = latest run. signals.csv = every run appended, so you keep a record."""
with open(stem + ".json", "w") as f:
json.dump(rows, f, indent=2)
path = stem + ".csv"
if os.path.exists(path):
with open(path, newline="") as f:
if next(csv.reader(f), None) != FIELDS:
raise SystemExit("%s exists with different columns (another bot?). Set OUT_STEM." % path)
new = not os.path.exists(path)
with open(path, "a", newline="") as f:
w = csv.DictWriter(f, FIELDS)
if new:
w.writeheader()
w.writerows(rows)
def main():
grades_on = os.environ.get("GRADES", "1") != "0"
locs = load(os.environ.get("LOCATIONS", "kalshi_locations.json"))
cost, cap = estimate_cost(locs, grades_on), float(os.environ.get("BUDGET_USD", "1.00"))
if cost > cap:
sys.exit("Estimated cost $%.4f is above BUDGET_USD $%.2f. Raise it on purpose or trim the list." % (cost, cap))
state_path = os.environ.get("STATE", "kalshi_state.json")
try:
with open(state_path) as f:
state = json.load(f)
except (OSError, ValueError):
state = {}
try:
rows = build(locs, state, grades_on=grades_on)
except gf.LowBalance as e:
sys.exit("PAUSED: %s" % e)
with open(state_path, "w") as f:
json.dump(state, f)
write(rows, os.environ.get("OUT_STEM", "signals"))
for r in rows: # short summary for your terminal
print("%-14s %s %s %s: now %s, margin %s, run-so-far %s, grade %s" % (
r["label"], r["metric"], r["op"], r["threshold"], r["value_now"],
r["margin_now"], r["extreme_so_far"], r["grade"]))
print("wrote %d rows (about $%.4f)" % (len(rows), cost))
if __name__ == "__main__":
main()