feat: add Gradio UI for viewing generated alphas with timestamps, copy button, and live generation

alpha_factory/ui.py

@@ -0,0 +1,180 @@
+"""
+Alpha Factory — Gradio UI
+View generated alphas, copy expressions, run new batches.
+
+Run: uv run python -m alpha_factory.ui
+"""
+import duckdb
+import gradio as gr
+from pathlib import Path
+from datetime import datetime
+
+
+DB_PATH = Path("factor_store/alphas.duckdb")
+
+
+def get_alphas_from_db(limit=50):
+    """Fetch alphas from DuckDB with all relevant info."""
+    if not DB_PATH.exists():
+        return []
+    conn = duckdb.connect(str(DB_PATH), read_only=True)
+    try:
+        rows = conn.execute(f"""
+            SELECT 
+                alpha_id,
+                submitted_at,
+                expression,
+                theme,
+                archetype,
+                anomaly_tag,
+                neutralization,
+                decay,
+                fields_used,
+                verdict
+            FROM alphas 
+            ORDER BY submitted_at DESC 
+            LIMIT {limit}
+        """).fetchall()
+        return rows
+    except Exception as e:
+        return []
+    finally:
+        conn.close()
+
+
+def format_alphas_table():
+    """Format alphas into a display-friendly list."""
+    rows = get_alphas_from_db()
+    if not rows:
+        return "No alphas generated yet. Run the pipeline first:\n\nuv run python -m alpha_factory.run --dry-run --batch-size 5"
+    
+    output = []
+    for row in rows:
+        alpha_id, submitted_at, expression, theme, archetype, tag, neutral, decay, fields, verdict = row
+        timestamp = submitted_at.strftime("%Y-%m-%d %H:%M:%S") if submitted_at else "unknown"
+        fields_str = ", ".join(fields) if fields else "none"
+        verdict_emoji = {"promote": "✅", "iterate": "🔄", "kill": "❌"}.get(verdict or "", "⏳")
+        
+        output.append(f"""{'='*80}
+{verdict_emoji} Alpha: {alpha_id}
+📅 Generated: {timestamp}
+🏷️ Theme: {theme} | Archetype: {archetype} | Tag: {tag}
+⚙️ Neutralization: {neutral} | Decay: {decay}
+📊 Fields: {fields_str}
+
+📝 EXPRESSION (copy below):
+{expression}
+{'='*80}
+""")
+    
+    return "\n".join(output)
+
+
+def get_single_alpha(index: int):
+    """Get a single alpha expression for easy copying."""
+    rows = get_alphas_from_db()
+    if not rows or index >= len(rows):
+        return "No alpha at this index"
+    return rows[index][2]  # expression column
+
+
+def get_alpha_cards():
+    """Return alpha data as list of dicts for dataframe display."""
+    rows = get_alphas_from_db()
+    if not rows:
+        return []
+    
+    data = []
+    for row in rows:
+        alpha_id, submitted_at, expression, theme, archetype, tag, neutral, decay, fields, verdict = row
+        timestamp = submitted_at.strftime("%Y-%m-%d %H:%M") if submitted_at else "?"
+        verdict_emoji = {"promote": "PASS", "iterate": "PENDING", "kill": "FAIL"}.get(verdict or "", "NEW")
+        data.append({
+            "Time": timestamp,
+            "ID": alpha_id[:8] + "...",
+            "Theme": theme or "",
+            "Archetype": archetype or "",
+            "Tag": tag or "",
+            "Decay": decay or 0,
+            "Status": verdict_emoji,
+            "Expression": expression[:100] + "..." if len(expression or "") > 100 else (expression or ""),
+        })
+    return data
+
+
+def get_full_expression(evt: gr.SelectData, data):
+    """When user clicks a row, show the full expression."""
+    if evt.index is None:
+        return ""
+    rows = get_alphas_from_db()
+    if not rows:
+        return ""
+    row_idx = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index
+    if row_idx < len(rows):
+        return rows[row_idx][2]  # full expression
+    return ""
+
+
+def run_batch_and_refresh(batch_size):
+    """Run a dry-run batch and refresh the table."""
+    import subprocess
+    import sys
+    result = subprocess.run(
+        [sys.executable, "-m", "alpha_factory.run", "--dry-run", "--batch-size", str(int(batch_size))],
+        capture_output=True, text=True, encoding="utf-8", errors="replace"
+    )
+    return get_alpha_cards(), result.stdout[-2000:] if result.stdout else result.stderr[-2000:]
+
+
+def build_ui():
+    """Build the Gradio interface."""
+    with gr.Blocks(title="Alpha Factory", theme=gr.themes.Soft()) as app:
+        gr.Markdown("""
+        # 🏭 Alpha Factory — Generated Alphas
+        View, copy, and manage alphas generated by the pipeline.
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=1):
+                batch_size_input = gr.Number(value=5, label="Batch Size", minimum=1, maximum=20)
+                generate_btn = gr.Button("🚀 Generate New Batch (Dry Run)", variant="primary")
+                refresh_btn = gr.Button("🔄 Refresh Table")
+            with gr.Column(scale=3):
+                gr.Markdown("### Stats")
+                stats_display = gr.Markdown("")
+        
+        gr.Markdown("### 📋 Generated Alphas (click row to see full expression)")
+        
+        alpha_table = gr.Dataframe(
+            value=get_alpha_cards(),
+            headers=["Time", "ID", "Theme", "Archetype", "Tag", "Decay", "Status", "Expression"],
+            interactive=False,
+            wrap=True,
+        )
+        
+        gr.Markdown("### 📝 Full Expression (click a row above)")
+        full_expr = gr.Textbox(
+            label="Full Expression (select & copy)",
+            lines=5,
+            show_copy_button=True,
+            interactive=False,
+        )
+        
+        gr.Markdown("### 📜 Pipeline Output")
+        pipeline_log = gr.Textbox(label="Last Run Log", lines=10, interactive=False)
+        
+        # Event handlers
+        alpha_table.select(get_full_expression, inputs=[alpha_table], outputs=[full_expr])
+        refresh_btn.click(lambda: get_alpha_cards(), outputs=[alpha_table])
+        generate_btn.click(
+            run_batch_and_refresh,
+            inputs=[batch_size_input],
+            outputs=[alpha_table, pipeline_log],
+        )
+    
+    return app
+
+
+if __name__ == "__main__":
+    app = build_ui()
+    app.launch(share=False)