-
Notifications
You must be signed in to change notification settings - Fork 256
Expand file tree
/
Copy pathbot_helpers.py
More file actions
149 lines (124 loc) · 4.69 KB
/
Copy pathbot_helpers.py
File metadata and controls
149 lines (124 loc) · 4.69 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
"""
Runtime helpers for the template bot: environment validation, startup/result
banners, and suppression of noisy upstream warnings.
Kept separate from main.py so that file can focus on the bot's forecasting
logic. main_with_no_framework.py keeps its own inline copies on purpose --
it's meant to be a single-file reference implementation.
"""
from __future__ import annotations
import logging
import os
import sys
import warnings
from typing import Any, Sequence
# Placeholder values shipped in .env.template. If a real env var still equals
# one of these the user forgot to replace it; we'd rather fail loudly here than
# inside the SDK three layers down.
_PLACEHOLDER_ENV_VALUES = {
"1234567890",
"REPLACE_ME",
"your-token-here",
"your-api-key-here",
}
def _is_real_env(name: str) -> bool:
val = os.getenv(name)
return bool(val and val.strip() and val.strip() not in _PLACEHOLDER_ENV_VALUES)
def silence_noisy_dependencies() -> None:
"""
Quiet warnings from transitive deps that fire on import and confuse new
users. Must be called *before* importing forecasting_tools.
"""
warnings.filterwarnings(
"ignore", message=r".*does not support cost tracking.*"
)
logging.getLogger("forecasting_tools.ai_models.model_tracker").setLevel(
logging.ERROR
)
# Streamlit installs its own logger hierarchy; suppress via its own API.
try:
from streamlit.logger import set_log_level
set_log_level("error")
except ImportError:
pass
# LiteLLM is verbose at INFO; its WARNING level is enough for us.
litellm_logger = logging.getLogger("LiteLLM")
litellm_logger.setLevel(logging.WARNING)
litellm_logger.propagate = False
def check_environment(strict: bool = True) -> None:
"""
Verify METACULUS_TOKEN is set; warn if no LLM key is configured. On
failure with strict=True, exits the process with a non-zero status.
"""
problems: list[str] = []
if not _is_real_env("METACULUS_TOKEN"):
problems.append(
"METACULUS_TOKEN is missing or still a placeholder. "
"Get one at https://www.metaculus.com/futureeval/participate/"
)
has_llm_key = any(
_is_real_env(k)
for k in ("OPENROUTER_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY")
)
if not has_llm_key:
print(
"⚠️ No LLM key set (OPENROUTER/OPENAI/ANTHROPIC). The bot will fall back\n"
" to the Metaculus LLM proxy. Free OpenRouter credits: "
"https://forms.gle/aQdYMq9Pisrf1v7d8\n"
)
if problems:
print("❌ Setup problems:")
for p in problems:
print(f" • {p}")
if strict:
sys.exit(1)
def print_startup_banner(run_mode: str, will_publish: bool) -> None:
publish = "publish=yes" if will_publish else "publish=no (dry run)"
print(f"🤖 Running mode={run_mode}, {publish}\n")
def print_run_summary_banner(
forecast_reports: Sequence[Any],
will_publish: bool,
tournament_url: str | None = None,
) -> None:
"""
End-of-run summary printed via print() (not logger) so it survives log
filtering. Shows count, per-question URLs, and any failure tracebacks.
If tournament_url is given, it's included as a footer link.
"""
# Lazy import so this module is usable in contexts where forecasting_tools
# isn't installed (e.g. unit tests of the banner format).
from forecasting_tools import ForecastReport
valid = [r for r in forecast_reports if isinstance(r, ForecastReport)]
exceptions = [r for r in forecast_reports if isinstance(r, BaseException)]
banner = "=" * 80
print()
print(banner)
if not forecast_reports:
print("ℹ️ No new questions to forecast on this run.")
print(banner)
print()
return
if valid and not exceptions:
verb = "submitted" if will_publish else "produced (dry run)"
print(f"🎉 Bot {verb} {len(valid)} forecast(s).")
elif valid and exceptions:
print(
f"⚠️ Partial — {len(valid)} succeeded, {len(exceptions)} failed."
)
else:
print(f"❌ All {len(exceptions)} attempt(s) failed.")
if valid:
print()
for r in valid:
note = f" (with {len(r.errors)} minor error(s))" if r.errors else ""
print(f" ✅ {r.question.page_url}{note}")
if will_publish and tournament_url:
print(f"\n Tournament: {tournament_url}")
if exceptions:
print()
for exc in exceptions:
msg = str(exc)
if len(msg) > 200:
msg = msg[:200] + "..."
print(f" ❌ {type(exc).__name__}: {msg}")
print(banner)
print()