from __future__ import annotations
import html
import os
from fastapi import APIRouter
from fastapi.responses import HTMLResponse
from app.config import settings
from database.connection import connect
from database.metrics_repository import fetch_resolved_rows
from database.resolution_repository import fetch_future_ohlcv
from evaluation.metrics import calculate_metrics
router = APIRouter(tags=["dashboard"])
@router.get("/resolution-status")
def resolution_status():
"""Return the latest durable resolution progress for dashboard polling."""
with connect(settings.DATABASE_URL) as conn, conn.cursor() as cur:
cur.execute(
"""
SELECT COUNT(*) AS total,
SUM(CASE WHEN resolved_at IS NOT NULL THEN 1 ELSE 0 END) AS resolved,
SUM(CASE WHEN resolved_at IS NULL THEN 1 ELSE 0 END) AS pending
FROM predictions
"""
)
counts = cur.fetchone()
cur.execute(
"""
SELECT status,
CAST(started_at AS TEXT) AS started_at,
CAST(finished_at AS TEXT) AS finished_at,
CAST(last_updated_at AS TEXT) AS last_updated_at,
total_rows, rows_processed, resolved_rows, failed_rows
FROM job_runs
WHERE job_name = 'resolution'
ORDER BY id DESC
LIMIT 1
"""
)
run = cur.fetchone()
cur.execute(
"""
SELECT MAX(CAST(value AS TEXT)) AS revision FROM (
SELECT MAX(CAST(prediction_timestamp AS TEXT)) AS value FROM predictions
UNION ALL SELECT MAX(CAST(resolved_at AS TEXT)) FROM predictions
UNION ALL SELECT MAX(CAST(calculated_at AS TEXT)) FROM prediction_metrics
UNION ALL SELECT MAX(CAST(COALESCE(last_updated_at, finished_at, started_at) AS TEXT)) FROM job_runs
)
"""
)
revision = cur.fetchone()["revision"]
total = int(counts["total"] or 0)
resolved = int(counts["resolved"] or 0)
pending = int(counts["pending"] or 0)
if run is None:
return {
"status": "IDLE",
"processed": 0,
"total": total,
"resolved": resolved,
"pending": pending,
"failed": 0,
"started_at": None,
"last_updated_at": None,
"revision": revision,
}
return {
"status": run["status"],
"processed": int(run["rows_processed"] or 0),
"total": int(run["total_rows"] or total),
"resolved": resolved,
"pending": pending,
"failed": int(run["failed_rows"] or 0),
"started_at": run["started_at"],
"last_updated_at": run["last_updated_at"] or run["finished_at"] or run["started_at"],
"revision": revision,
}
PREDICTION_COLUMNS = """
id,
prediction_date,
symbol,
predicted_probability,
rank,
prediction_close,
target_threshold,
target_horizon_days,
entry_date,
entry_open,
max_close_5d,
evaluation_end_date,
actual_return,
actual_label,
resolved_at
"""
# ─────────────────────────────────────────────
# Data access
# ─────────────────────────────────────────────
def _fetch_predictions(
conn,
*,
resolved: bool | None = None,
limit: int = 1000,
):
where = ""
if resolved is True:
where = "WHERE actual_label IS NOT NULL"
elif resolved is False:
where = "WHERE actual_label IS NULL"
with conn.cursor() as cur:
cur.execute(
f"""
SELECT {PREDICTION_COLUMNS}
FROM predictions
{where}
ORDER BY prediction_date DESC, rank ASC
LIMIT ?
""",
(limit,),
)
return cur.fetchall()
def _fetch_prediction(conn, prediction_id: int):
with conn.cursor() as cur:
cur.execute(
f"""
SELECT {PREDICTION_COLUMNS}
FROM predictions
WHERE id = ?
""",
(prediction_id,),
)
return cur.fetchone()
def _fetch_default_prediction(conn):
with conn.cursor() as cur:
cur.execute(
f"""
SELECT {PREDICTION_COLUMNS}
FROM predictions
ORDER BY prediction_date DESC, rank ASC
LIMIT 1
"""
)
return cur.fetchone()
def _select_prediction(conn, prediction_id: int | None):
if prediction_id is not None:
row = _fetch_prediction(conn, prediction_id)
if row is not None:
return row
return _fetch_default_prediction(conn)
# ─────────────────────────────────────────────
# Small SVG primitives
# ─────────────────────────────────────────────
def _rounded_top_rect(
x: float,
y: float,
w: float,
h: float,
r: float = 4,
) -> str:
if h <= 0:
return ""
r = min(r, h, w / 2)
bottom = y + h
return (
f''
)
def _empty_chart_message(
width: int,
height: int,
message: str,
) -> str:
return (
f''
)
# ─────────────────────────────────────────────
# Calibration
# ─────────────────────────────────────────────
def _build_calibration_buckets(rows) -> list[dict]:
"""
Build 10 probability buckets.
D1 = 0-10%
D2 = 10-20%
...
D10 = 90-100%
For each bucket:
count
mean predicted probability
actual positive rate
"""
buckets = [
{
"decile": i + 1,
"count": 0,
"predicted_probability": None,
"positive_rate": None,
}
for i in range(10)
]
for row in rows:
probability = row["predicted_probability"]
actual_label = row["actual_label"]
if probability is None or actual_label is None:
continue
probability = float(probability)
# Protect against numerical values slightly outside [0, 1].
probability = max(0.0, min(1.0, probability))
index = min(int(probability * 10), 9)
bucket = buckets[index]
bucket["count"] += 1
if bucket["predicted_probability"] is None:
bucket["predicted_probability"] = []
bucket["predicted_probability"].append(probability)
if "positives" not in bucket:
bucket["positives"] = 0
bucket["positives"] += int(actual_label)
for bucket in buckets:
count = bucket["count"]
if count:
probabilities = bucket["predicted_probability"]
bucket["predicted_probability"] = (
sum(probabilities) / len(probabilities)
)
bucket["positive_rate"] = (
bucket["positives"] / count
)
else:
bucket["predicted_probability"] = None
bucket["positive_rate"] = None
bucket.pop("positives", None)
return buckets
def _svg_calibration_chart(
deciles: list[dict],
overall_hit_rate: float | None,
) -> str:
if not deciles or not any(d["count"] for d in deciles):
return _empty_chart_message(
720,
300,
"No resolved predictions available for calibration.",
)
width, height = 720, 300
margin_left = 50
margin_right = 20
margin_top = 30
margin_bottom = 55
plot_w = width - margin_left - margin_right
plot_h = height - margin_top - margin_bottom
gap = 8
bar_w = (
plot_w - gap * (len(deciles) - 1)
) / len(deciles)
def y_of(rate: float) -> float:
return (
margin_top
+ plot_h
- rate * plot_h
)
parts = [
f'")
return "".join(parts)
# ─────────────────────────────────────────────
# Price trajectory
# ─────────────────────────────────────────────
def _svg_price_trajectory(
prediction: dict,
ohlcv_rows: list[dict],
) -> str:
if not ohlcv_rows:
return _empty_chart_message(
640,
220,
"Awaiting D+1 market data for this prediction.",
)
width, height = 640, 220
margin_left = 48
margin_right = 16
margin_top = 20
margin_bottom = 32
plot_w = width - margin_left - margin_right
plot_h = height - margin_top - margin_bottom
entry_open = ohlcv_rows[0]["open"]
target_close = (
entry_open
* (1 + prediction["target_threshold"])
)
labels = (
["D"]
+ [
f"D+{i + 1}"
for i in range(len(ohlcv_rows))
]
)
values = (
[prediction["prediction_close"]]
+ [row["close"] for row in ohlcv_rows]
)
lo = min(values + [target_close, entry_open])
hi = max(values + [target_close, entry_open])
pad = (hi - lo) * 0.12 or 1.0
lo -= pad
hi += pad
n = len(labels)
step = plot_w / max(n - 1, 1)
def x_of(i: int) -> float:
return margin_left + i * step
def y_of(v: float) -> float:
return (
margin_top
+ plot_h
- (v - lo) / (hi - lo) * plot_h
)
parts = [
f'")
return "".join(parts)
# ─────────────────────────────────────────────
# Table rendering
# ─────────────────────────────────────────────
def _outcome_style(actual_label) -> str:
if actual_label is None:
return "color:var(--text-muted)"
if int(actual_label) == 1:
return "color:var(--good);font-weight:600"
return "color:var(--critical);font-weight:600"
def _format_value(value, digits: int | None = None):
if value is None:
return "—"
if digits is not None:
try:
return f"{float(value):.{digits}f}"
except (TypeError, ValueError):
pass
return str(value)
def _render_table(
rows,
*,
title: str,
empty_message: str,
) -> str:
columns = [
("prediction_date", "Date"),
("symbol", "Symbol"),
("predicted_probability", "Probability"),
("rank", "Rank"),
("prediction_close", "Prediction close"),
("entry_date", "Entry date"),
("entry_open", "Entry open"),
("target_close", "Target close"),
("max_close_5d", "Max close 5D"),
("actual_return", "Actual return"),
("actual_label", "Outcome"),
("resolved_at", "Resolved"),
]
head = "".join(
f"
{html.escape(label)}
"
for _, label in columns
)
if not rows:
return f"""
{html.escape(title)}
0
{html.escape(empty_message)}
"""
body_rows = []
for row in rows:
target_close = None
if row["entry_open"] is not None:
target_close = (
float(row["entry_open"])
* (1 + float(row["target_threshold"]))
)
cells = []
for col, _label in columns:
if col == "prediction_date":
value = (
f''
f'{html.escape(str(row["prediction_date"]))}'
f''
)
elif col == "symbol":
value = html.escape(str(row["symbol"]))
elif col == "predicted_probability":
value = (
_format_value(
float(row[col]) * 100,
1,
)
+ "%"
if row[col] is not None
else "—"
)
elif col in {
"prediction_close",
"entry_open",
"target_close",
"max_close_5d",
}:
value = _format_value(
target_close
if col == "target_close"
else row[col],
2,
)
elif col == "actual_return":
value = (
_format_value(
float(row[col]) * 100,
2,
)
+ "%"
if row[col] is not None
else "—"
)
elif col == "actual_label":
if row[col] is None:
label = "pending"
elif int(row[col]) == 1:
label = "hit"
else:
label = "miss"
value = (
f''
f'{label}'
f''
)
else:
value = html.escape(
_format_value(row[col])
)
cells.append(f"
{value}
")
body_rows.append(
"
" + "".join(cells) + "
"
)
return f"""
{html.escape(title)}
{len(rows)}
{head}
{"".join(body_rows)}
"""
# ─────────────────────────────────────────────
# Route
# ─────────────────────────────────────────────
@router.get(
"/dashboard",
response_class=HTMLResponse,
)
def dashboard(prediction_id: int | None = None):
threshold = float(
os.getenv("PREDICTION_THRESHOLD", "0.5")
)
with connect(settings.DATABASE_URL) as conn:
# Selected prediction.
selected = _select_prediction(
conn,
prediction_id,
)
trajectory_rows = []
if selected is not None:
trajectory_rows = fetch_future_ohlcv(
conn,
selected["symbol"],
selected["prediction_date"],
selected["target_horizon_days"],
)
# ALL resolved predictions.
resolved_rows = _fetch_predictions(
conn,
resolved=True,
limit=5000,
)
# ALL unresolved/current predictions.
pending_rows = _fetch_predictions(
conn,
resolved=False,
limit=1000,
)
# Metrics/calibration source.
metrics_rows = fetch_resolved_rows(
conn,
days=None,
)
# ─────────────────────────────────────────
# Calibration
# ─────────────────────────────────────────
calibration_rows = [
row
for row in resolved_rows
if row["predicted_probability"] is not None
and row["actual_label"] is not None
]
deciles = _build_calibration_buckets(
calibration_rows
)
if calibration_rows:
overall_hit_rate = sum(
int(row["actual_label"])
for row in calibration_rows
) / len(calibration_rows)
else:
overall_hit_rate = None
calibration_svg = _svg_calibration_chart(
deciles,
overall_hit_rate,
)
# ─────────────────────────────────────────
# Selected prediction chart
# ─────────────────────────────────────────
if selected is not None:
trajectory_svg = _svg_price_trajectory(
selected,
trajectory_rows,
)
trajectory_heading = (
f'{html.escape(selected["symbol"])} '
f'— predicted '
f'{html.escape(str(selected["prediction_date"]))} '
f'(probability '
f'{float(selected["predicted_probability"]) * 100:.1f}%)'
)
else:
trajectory_svg = _empty_chart_message(
640,
220,
"No predictions yet.",
)
trajectory_heading = (
"No prediction selected"
)
# ─────────────────────────────────────────
# Metrics
# ─────────────────────────────────────────
if metrics_rows:
metrics_result = calculate_metrics(
metrics_rows,
threshold=threshold,
)
else:
metrics_result = None
if metrics_result is not None:
total_resolved = len(metrics_rows)
hit_rate = (
metrics_result.hit_rate
if metrics_result.hit_rate is not None
else 0
)
metrics_html = f"""
Resolved
{total_resolved}
Hit rate
{hit_rate * 100:.1f}%
Prediction threshold
{threshold * 100:.0f}%
Pending
{len(pending_rows)}
"""
else:
metrics_html = """
Resolved
0
Pending
0
"""
# ─────────────────────────────────────────
# Tables
# ─────────────────────────────────────────
resolved_table = _render_table(
resolved_rows,
title="Resolved predictions",
empty_message=(
"No predictions have been resolved yet."
),
)
pending_table = _render_table(
pending_rows,
title="Current predictions",
empty_message=(
"There are no unresolved predictions."
),
)
return HTMLResponse(
f"""
Stock signal dashboard
Stock signal dashboard
Predictions, five-session outcomes, calibration,
and currently unresolved signals.
Prediction resolution
Loading
0 / 0 processed
0.00%
Resolved 0
Pending 0
Failed 0
Started —
Last update —
{metrics_html}
Calibration
Does a higher predicted probability actually correspond
to a higher probability of hitting the target?