-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
108 lines (88 loc) · 3.05 KB
/
Copy pathmain.py
File metadata and controls
108 lines (88 loc) · 3.05 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
"""OpenClaw skill entrypoint for Kalshi market data.
This module wraps scripts/kalshi_api.py so OpenClaw can load and
execute it as a plugin. Uses the Kalshi public REST API — no
external dependencies required.
"""
import json
import sys
import os
# Add the skill root to sys.path so we can import the API client
_skill_root = os.path.abspath(os.path.dirname(__file__))
if _skill_root not in sys.path:
sys.path.insert(0, _skill_root)
from scripts.kalshi_api import (
parse_kalshi_url,
fetch_events,
fetch_event,
events_to_browse_list,
event_to_market_result,
KalshiAPIError,
)
def browse(url: str = "https://kalshi.com", max_markets: int = 20) -> list[dict]:
"""List markets from a Kalshi category or home page.
Args:
url: Kalshi page URL or category URL.
max_markets: Maximum number of markets to return.
Returns:
List of dicts with 'title' and 'url' keys.
"""
parsed = parse_kalshi_url(url)
category_slug = parsed.get("category_slug")
events = fetch_events(category_slug=category_slug, limit=max_markets)
return events_to_browse_list(events, max_markets=max_markets)
def market(url: str) -> dict:
"""Fetch data for a single Kalshi market/event.
Args:
url: Full URL of the Kalshi market page, or an event ticker.
Returns:
Dict with 'title', 'outcomes', 'status', and 'error' keys.
"""
parsed = parse_kalshi_url(url)
event_ticker = parsed.get("event_ticker")
if not event_ticker:
return {
"title": None,
"outcomes": [],
"status": "error",
"error": f"Could not extract event ticker from URL: {url}",
}
try:
event = fetch_event(event_ticker)
return event_to_market_result(event)
except KalshiAPIError as exc:
return {
"title": None,
"outcomes": [],
"status": "error",
"error": str(exc),
}
# Allow direct execution: python main.py browse|market [args]
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python main.py browse [--url URL] [--max N]")
print(" python main.py market <URL>")
sys.exit(1)
command = sys.argv[1]
if command == "browse":
url = "https://kalshi.com"
max_m = 20
args = sys.argv[2:]
i = 0
while i < len(args):
if args[i] == "--url" and i + 1 < len(args):
url = args[i + 1]
i += 2
elif args[i] == "--max" and i + 1 < len(args):
max_m = int(args[i + 1])
i += 2
else:
i += 1
print(json.dumps(browse(url=url, max_markets=max_m), indent=2, ensure_ascii=False))
elif command == "market":
if len(sys.argv) < 3:
print("Usage: python main.py market <URL>", file=sys.stderr)
sys.exit(1)
print(json.dumps(market(url=sys.argv[2]), indent=2, ensure_ascii=False))
else:
print(f"Unknown command: {command}", file=sys.stderr)
sys.exit(1)