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
|
#!/usr/bin/env python3
"""
slopbot - Zulip bot that replies with stock prices when it sees $TICKER patterns.
Usage:
python main.py
Requires a .zuliprc file in the same directory (see .zuliprc.example).
"""
import logging
import re
import zulip
import yfinance as yf
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("slopbot")
# Regex to match stock tickers like $NVDA, $AAPL, $BRK
TICKER_RE = re.compile(r'\$([A-Z]{1,5})')
log.info("Connecting to Zulip...")
client = zulip.Client(config_file='.zuliprc')
log.info("Connected as %s", client.email)
def fetch_price(ticker: str) -> str:
"""Return a formatted price string for a ticker, or an error message."""
log.info("Fetching price for %s", ticker)
try:
price = yf.Ticker(ticker).fast_info['lastPrice']
result = f"${ticker}: ${price:,.2f}"
log.info(" -> %s", result)
return result
except Exception as e:
log.warning(" -> %s unavailable: %s", ticker, e)
return f"${ticker}: unavailable ({e})"
def handle_event(event: dict) -> None:
"""Process a Zulip event and reply if tickers are found."""
if event.get("type") != "message":
return
msg = event["message"]
# Don't reply to ourselves
if msg["sender_email"] == client.email:
return
content = msg.get("content", "")
tickers = TICKER_RE.findall(content.upper())
if not tickers:
return
log.info(
"Message from %s in %s/%s: found tickers %s",
msg["sender_email"],
msg.get("display_recipient", "?"),
msg.get("subject", "?"),
tickers,
)
lines = [fetch_price(t) for t in dict.fromkeys(tickers)] # dedup, preserve order
reply = "\n".join(lines)
client.send_message({
"type": msg["type"],
"to": msg["stream_id"] if msg["type"] == "stream" else msg["sender_email"],
"topic": msg.get("subject", ""),
"content": reply,
})
log.info("Reply sent.")
log.info("Listening for messages...")
client.call_on_each_event(handle_event, event_types=["message"])
|
