feat(hypotheses): add Hypotheses dashboard tab

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Youssef Aitousarrah 2026-04-10 09:52:58 -07:00
parent fe5b8886c0
commit 5b87a56f31
4 changed files with 253 additions and 1 deletions

View File

@ -0,0 +1,73 @@
"""Tests for the hypotheses dashboard page data loading."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from tradingagents.ui.pages.hypotheses import (
load_active_hypotheses,
load_concluded_hypotheses,
days_until_ready,
)
def test_load_active_hypotheses(tmp_path):
active = {
"max_active": 5,
"hypotheses": [
{
"id": "options_flow-test",
"title": "Test hypothesis",
"scanner": "options_flow",
"status": "running",
"priority": 7,
"days_elapsed": 5,
"min_days": 14,
"created_at": "2026-04-01",
"picks_log": ["2026-04-01"] * 5,
"conclusion": None,
}
],
}
f = tmp_path / "active.json"
f.write_text(json.dumps(active))
result = load_active_hypotheses(str(f))
assert len(result) == 1
assert result[0]["id"] == "options_flow-test"
def test_load_active_hypotheses_missing_file(tmp_path):
result = load_active_hypotheses(str(tmp_path / "missing.json"))
assert result == []
def test_load_concluded_hypotheses(tmp_path):
doc = tmp_path / "2026-04-10-options_flow-test.md"
doc.write_text(
"# Hypothesis: Test\n\n"
"**Scanner:** options_flow\n"
"**Period:** 2026-03-27 → 2026-04-10 (14 days)\n"
"**Outcome:** accepted ✅\n"
)
results = load_concluded_hypotheses(str(tmp_path))
assert len(results) == 1
assert results[0]["filename"] == doc.name
assert results[0]["outcome"] == "accepted ✅"
def test_load_concluded_hypotheses_empty_dir(tmp_path):
results = load_concluded_hypotheses(str(tmp_path))
assert results == []
def test_days_until_ready_has_days_left():
hyp = {"days_elapsed": 5, "min_days": 14}
assert days_until_ready(hyp) == 9
def test_days_until_ready_past_due():
hyp = {"days_elapsed": 15, "min_days": 14}
assert days_until_ready(hyp) == 0

View File

@ -52,7 +52,7 @@ def render_sidebar():
# Navigation
page = st.radio(
"Navigation",
options=["Overview", "Signals", "Portfolio", "Performance", "Config"],
options=["Overview", "Signals", "Portfolio", "Performance", "Hypotheses", "Config"],
label_visibility="collapsed",
)
@ -116,6 +116,7 @@ def route_page(page):
"Signals": pages.todays_picks,
"Portfolio": pages.portfolio,
"Performance": pages.performance,
"Hypotheses": pages.hypotheses,
"Config": pages.settings,
}
module = page_map.get(page)

View File

@ -39,6 +39,12 @@ except Exception as _e:
_logger.error("Failed to import settings page: %s", _e, exc_info=True)
settings = None
try:
from tradingagents.ui.pages import hypotheses
except Exception as _e:
_logger.error("Failed to import hypotheses page: %s", _e, exc_info=True)
hypotheses = None
__all__ = [
"home",
@ -46,4 +52,5 @@ __all__ = [
"portfolio",
"performance",
"settings",
"hypotheses",
]

View File

@ -0,0 +1,171 @@
"""
Hypotheses dashboard page tracks active and concluded experiments.
Reads docs/iterations/hypotheses/active.json and the concluded/ directory.
No external API calls; all data is file-based.
"""
import json
import re
from pathlib import Path
from typing import Any, Dict, List
import streamlit as st
from tradingagents.ui.theme import COLORS, page_header
_REPO_ROOT = Path(__file__).parent.parent.parent.parent
_ACTIVE_JSON = _REPO_ROOT / "docs/iterations/hypotheses/active.json"
_CONCLUDED_DIR = _REPO_ROOT / "docs/iterations/hypotheses/concluded"
def load_active_hypotheses(active_path: str = str(_ACTIVE_JSON)) -> List[Dict[str, Any]]:
"""Load all hypotheses from active.json. Returns [] if file missing."""
path = Path(active_path)
if not path.exists():
return []
try:
with open(path) as f:
data = json.load(f)
return data.get("hypotheses", [])
except Exception:
return []
def load_concluded_hypotheses(concluded_dir: str = str(_CONCLUDED_DIR)) -> List[Dict[str, Any]]:
"""
Load concluded hypothesis metadata by parsing markdown files in concluded/.
Extracts: filename, title, scanner, period, outcome.
"""
dir_path = Path(concluded_dir)
if not dir_path.exists():
return []
results = []
for md_file in sorted(dir_path.glob("*.md"), reverse=True):
if md_file.name == ".gitkeep":
continue
try:
text = md_file.read_text()
title = _extract_md_field(text, r"^# Hypothesis: (.+)$")
scanner = _extract_md_field(text, r"^\*\*Scanner:\*\* (.+)$")
period = _extract_md_field(text, r"^\*\*Period:\*\* (.+)$")
outcome = _extract_md_field(text, r"^\*\*Outcome:\*\* (.+)$")
results.append({
"filename": md_file.name,
"title": title or md_file.stem,
"scanner": scanner or "",
"period": period or "",
"outcome": outcome or "",
})
except Exception:
continue
return results
def _extract_md_field(text: str, pattern: str) -> str:
"""Extract a field value from a markdown line using regex."""
match = re.search(pattern, text, re.MULTILINE)
return match.group(1).strip() if match else ""
def days_until_ready(hyp: Dict[str, Any]) -> int:
"""Return number of days remaining before hypothesis can conclude (min 0)."""
return max(0, hyp.get("min_days", 14) - hyp.get("days_elapsed", 0))
def render() -> None:
"""Render the hypotheses tracking page."""
st.markdown(
page_header("Hypotheses", "Active experiments & concluded findings"),
unsafe_allow_html=True,
)
hypotheses = load_active_hypotheses()
concluded = load_concluded_hypotheses()
if not hypotheses and not concluded:
st.info(
"No hypotheses yet. Run `/backtest-hypothesis \"<description>\"` to start an experiment."
)
return
running = [h for h in hypotheses if h["status"] == "running"]
pending = [h for h in hypotheses if h["status"] == "pending"]
st.markdown(
f'<div class="section-title">Active Experiments '
f'<span class="accent">// {len(running)} running, {len(pending)} pending</span></div>',
unsafe_allow_html=True,
)
if running or pending:
import pandas as pd
active_rows = []
for h in sorted(running + pending, key=lambda x: -x.get("priority", 0)):
days_left = days_until_ready(h)
ready_str = "concluding soon" if days_left == 0 else f"{days_left}d left"
active_rows.append({
"ID": h["id"],
"Title": h.get("title", ""),
"Scanner": h.get("scanner", ""),
"Status": h["status"],
"Progress": f"{h.get('days_elapsed', 0)}/{h.get('min_days', 14)}d",
"Picks": len(h.get("picks_log", [])),
"Ready": ready_str,
"Priority": h.get("priority", ""),
})
df = pd.DataFrame(active_rows)
st.dataframe(
df,
width="stretch",
hide_index=True,
column_config={
"ID": st.column_config.TextColumn(width="medium"),
"Title": st.column_config.TextColumn(width="large"),
"Scanner": st.column_config.TextColumn(width="medium"),
"Status": st.column_config.TextColumn(width="small"),
"Progress": st.column_config.TextColumn(width="small"),
"Picks": st.column_config.NumberColumn(format="%d", width="small"),
"Ready": st.column_config.TextColumn(width="medium"),
"Priority": st.column_config.NumberColumn(format="%d/9", width="small"),
},
)
else:
st.info("No active experiments.")
st.markdown("<div style='height:1.5rem;'></div>", unsafe_allow_html=True)
st.markdown(
f'<div class="section-title">Concluded Experiments '
f'<span class="accent">// {len(concluded)} total</span></div>',
unsafe_allow_html=True,
)
if concluded:
import pandas as pd
concluded_rows = []
for c in concluded:
outcome = c["outcome"]
emoji = "" if "accepted" in outcome else ""
concluded_rows.append({
"Date": c["filename"][:10],
"Title": c["title"],
"Scanner": c["scanner"],
"Period": c["period"],
"Outcome": emoji,
})
cdf = pd.DataFrame(concluded_rows)
st.dataframe(
cdf,
width="stretch",
hide_index=True,
column_config={
"Date": st.column_config.TextColumn(width="small"),
"Title": st.column_config.TextColumn(width="large"),
"Scanner": st.column_config.TextColumn(width="medium"),
"Period": st.column_config.TextColumn(width="medium"),
"Outcome": st.column_config.TextColumn(width="small"),
},
)
else:
st.info("No concluded experiments yet.")