""" Lightweight module wrappers for local asset modules. Provides safe, minimal wrappers with standardized init() and render(container) API. Wrappers included: - HTMLModule: render local .html files (embedded via components.html) - VolRegimeModule: import vol-regime-prediction-main (if present) and expose run_eda if available; shows pre-generated figures - CrashMonitorModule: use crash_monitor data fetcher/calculator to produce quick charts (uses synthetic demo data to avoid blocking external calls) This file avoids heavy imports at top-level; GUI libraries are imported inside render() methods. """ from __future__ import annotations import os import sys import traceback from pathlib import Path from typing import Optional class BaseModule: def __init__(self, name: str, path: str): self.name = name self.path = Path(path) self.available = True self._error: Optional[str] = None def init(self) -> None: """Initialize the module (no-op by default).""" return None def render(self, container) -> None: """Render the module into the provided Streamlit container. The container is expected to be a Streamlit DeltaGenerator used via "with container:" in the caller. """ raise NotImplementedError class HTMLModule(BaseModule): def init(self) -> None: if not self.path.exists(): self.available = False self._error = f"HTML file not found: {self.path}" def render(self, container) -> None: import streamlit as st import streamlit.components.v1 as components with container: if not self.available: st.error(self._error or "HTML module not available") return try: html = self.path.read_text(encoding="utf-8") except Exception as e: st.error(f"Failed to read HTML: {e}") return # Inline the HTML in an iframe-like component. Height is conservative. components.html(html, height=700, scrolling=True) class VolRegimeModule(BaseModule): def init(self) -> None: # We try to import the project's analysis module by adding the project # root (folder that contains src/) to sys.path. try: project_root = self.path if not project_root.exists(): raise FileNotFoundError(f"Vol-regime project root not found: {project_root}") # Add project root to sys.path to allow `import src...` pr = str(project_root) if pr not in sys.path: sys.path.insert(0, pr) # Import lazily try: import src.analysis.eda as eda # type: ignore self._eda = eda # main visualization entrypoints we might call self.main_fn = getattr(eda, "run_eda", None) self.available = True except Exception as e: # don't fail hard — we'll fallback to showing pre-built figures self._eda = None self.main_fn = None self.available = False self._error = str(e) except Exception as e: self.available = False self._error = str(e) def render(self, container) -> None: import streamlit as st with container: st.header("Volatility Regime — Vol Regime Prediction") if self.available and getattr(self, "_eda", None) is not None: st.success("vol-regime module imported successfully") # Show quick access to main function if available if self.main_fn is not None: if st.button("Run vol-regime main visualization (run_eda)"): try: with st.spinner("Running run_eda() — may take a while..."): # run_eda writes figures to reports/figures by default out = self.main_fn() st.success("run_eda() completed") except Exception as e: st.exception(f"run_eda() failed: {e}") # Show pre-generated figures if present figs_dir = self.path / "reports" / "figures" if figs_dir.exists(): imgs = sorted([p for p in figs_dir.glob("*.png")]) if imgs: st.markdown("**Pre-generated figures**") for img in imgs[:6]: st.image(str(img), use_column_width=True) else: st.info("No pre-generated figures found in reports/figures") else: st.info("No reports/figures directory found — try running the module's run_eda() from its project root") else: st.warning("vol-regime module not importable: " + (self._error or "unknown")) # Also try to show any existing static figures even if import failed figs_dir = self.path / "reports" / "figures" if figs_dir.exists(): imgs = sorted([p for p in figs_dir.glob("*.png")]) if imgs: st.markdown("**Pre-generated figures (import failed, showing static files)**") for img in imgs[:6]: st.image(str(img), use_column_width=True) class CrashMonitorModule(BaseModule): def init(self) -> None: # Import crash_monitor helper modules directly from files to avoid # package-level side-effects (e.g., flask/blueprint imports). try: fetcher_path = self.path / 'data' / 'fetcher.py' calc_path = self.path / 'data' / 'calculator.py' if not fetcher_path.exists() or not calc_path.exists(): raise FileNotFoundError(f"crash_monitor data modules not found under {self.path}") import importlib.util spec_f = importlib.util.spec_from_file_location(f"crash_monitor_fetcher_{self.name}", str(fetcher_path)) fetcher_mod = importlib.util.module_from_spec(spec_f) spec_f.loader.exec_module(fetcher_mod) spec_c = importlib.util.spec_from_file_location(f"crash_monitor_calc_{self.name}", str(calc_path)) calc_mod = importlib.util.module_from_spec(spec_c) spec_c.loader.exec_module(calc_mod) self.fetcher = fetcher_mod self.calc = calc_mod self.available = True except Exception as e: self.fetcher = None self.calc = None self.available = False self._error = str(e) def render(self, container) -> None: import streamlit as st import pandas as pd import numpy as np import matplotlib.pyplot as plt with container: st.header("Crash Monitor — Quick Charts") if not (getattr(self, "fetcher", None) and getattr(self, "calc", None)): st.warning("Crash monitor not available: " + (self._error or "unknown")) return try: # Instantiate DataFetcher but avoid slow external sources — use synthetic demo DataFetcher = getattr(self.fetcher, "DataFetcher") dfetch = DataFetcher() try: # Fast synthetic demo path (guaranteed local) src = dfetch._synthetic_demo("SPX") except Exception: # Fallback to safe fetch (may still be networked) src = dfetch.fetch_options_chain("SPX") spot = getattr(src, "spot", None) or src[0] if isinstance(src, (list, tuple)) else None # src is SourceResult dataclass; it has .strikes and .spot strikes = getattr(src, "strikes", None) or [] if not strikes: st.info("No strike data available from crash monitor (empty).") return # Crash profile profile = self.calc.compute_crash_profile(strikes, spot) dfp = pd.DataFrame(profile).sort_values("spot_pct") dfp = dfp.set_index("spot_pct") st.markdown("**GEX+ vs Spot % (Crash Profile)**") st.line_chart(dfp["gex_plus"]) # Heatmap grid = self.calc.compute_heatmap_grid(strikes, spot) values = np.array(grid.get("values", [])) if values.size: fig, ax = plt.subplots(figsize=(8, 4)) im = ax.imshow(values, aspect="auto", origin="lower", cmap="RdBu_r", vmin=-1, vmax=1) ax.set_title("GEX+ heatmap (normalized)") ax.set_xlabel("IV shock bins") ax.set_ylabel("Spot % bins") plt.colorbar(im, ax=ax, fraction=0.046, pad=0.04) st.pyplot(fig) else: st.info("Heatmap not available") except Exception as e: st.exception(f"Crash monitor render error: {e}\n" + traceback.format_exc()) # Helper factory functions def _html_module(name: str, path: str) -> HTMLModule: return HTMLModule(name, path) def _vol_module(name: str, path: str) -> VolRegimeModule: return VolRegimeModule(name, path) def _crash_module(name: str, path: str) -> CrashMonitorModule: return CrashMonitorModule(name, path) # Public API — simple classes for callers to instantiate HTMLModule = HTMLModule VolRegimeModule = VolRegimeModule CrashMonitorModule = CrashMonitorModule