"""Position page — drilldown on a single open or recently-closed trade.""" from __future__ import annotations import json import os from pathlib import Path from uuid import UUID import plotly.graph_objects as go import streamlit as st from cerbero_bite.gui.data_layer import ( DEFAULT_DB_PATH, compute_distance_metrics, compute_payoff_curve, humanize_dt, load_closed_positions, load_decisions_for_position, load_open_positions, load_position_by_id, ) from cerbero_bite.state.models import PositionRecord def _resolve_db() -> Path: return Path(os.environ.get("CERBERO_BITE_GUI_DB", DEFAULT_DB_PATH)) def _position_label(p: PositionRecord) -> str: short = ( f"{int(p.short_strike)}/{int(p.long_strike)}" if p.short_strike and p.long_strike else "—" ) return f"{str(p.proposal_id)[:8]} · {p.spread_type} · {short} · {p.status}" def _render_header(position: PositionRecord) -> None: cols = st.columns(4) cols[0].metric("stato", position.status) cols[1].metric("spread", position.spread_type) cols[2].metric("contratti", position.n_contracts) cols[3].metric("credito (USD)", f"${float(position.credit_usd):+.2f}") st.caption( f"`{position.proposal_id}` · aperta il " f"{humanize_dt(position.opened_at)} · scadenza " f"{humanize_dt(position.expiry)}" ) def _render_legs(position: PositionRecord) -> None: st.subheader("Gambe (snapshot all'entrata)") rows = [ { "gamba": "short", "strumento": position.short_instrument, "strike": float(position.short_strike), "lato": "VENDI", "size": position.n_contracts, "delta all'entrata": float(position.delta_at_entry), }, { "gamba": "long", "strumento": position.long_instrument, "strike": float(position.long_strike), "lato": "COMPRA", "size": position.n_contracts, "delta all'entrata": "—", }, ] st.dataframe(rows, use_container_width=True, hide_index=True) st.caption( "Mid e greche live non vengono richiesti agli MCP dal cruscotto. " "Il refresh è demandato al motore: visibile nella pagina Audit." ) def _render_distance(position: PositionRecord) -> None: metrics = compute_distance_metrics(position) cols = st.columns(5) cols[0].metric( "Short OTM %", f"{metrics.short_strike_otm_pct:.1%}" if metrics.short_strike_otm_pct is not None else "—", ) cols[1].metric( "Giorni a scadenza", metrics.days_to_expiry if metrics.days_to_expiry is not None else "—", ) cols[2].metric( "Giorni in tenuta", metrics.days_held if metrics.days_held is not None else "—", ) cols[3].metric("Δ all'entrata", f"{metrics.delta_at_entry:+.3f}") cols[4].metric("Larghezza % spot", f"{metrics.width_pct_of_spot:.1%}") def _render_payoff(position: PositionRecord) -> None: st.subheader("Payoff a scadenza") curve = compute_payoff_curve(position) fig = go.Figure() fig.add_trace( go.Scatter( x=curve.spot_grid, y=curve.pnl_grid_usd, mode="lines", line={"color": "#3498db", "width": 2.5}, name="P&L a scadenza", fill="tozeroy", fillcolor="rgba(52,152,219,0.10)", ) ) fig.add_hline(y=0, line_dash="dot", line_color="grey", opacity=0.5) fig.add_vline( x=curve.short_strike, line_dash="dash", line_color="#27ae60", opacity=0.7, annotation_text=f"short {curve.short_strike:.0f}", annotation_position="top", ) fig.add_vline( x=curve.long_strike, line_dash="dash", line_color="#c0392b", opacity=0.7, annotation_text=f"long {curve.long_strike:.0f}", annotation_position="top", ) if curve.breakeven is not None: fig.add_vline( x=curve.breakeven, line_dash="dot", line_color="orange", opacity=0.7, annotation_text=f"BE {curve.breakeven:.2f}", annotation_position="bottom", ) fig.add_vline( x=curve.spot_at_entry, line_dash="solid", line_color="#7f8c8d", opacity=0.4, annotation_text=f"spot all'entrata {curve.spot_at_entry:.0f}", annotation_position="bottom", ) fig.update_layout( height=380, margin={"l": 10, "r": 10, "t": 30, "b": 10}, xaxis_title="ETH spot a scadenza (USD)", yaxis_title="P&L (USD)", legend={"orientation": "h", "y": 1.1}, ) st.plotly_chart(fig, use_container_width=True) cols = st.columns(3) cols[0].metric("Profitto massimo", f"${curve.max_profit_usd:+.2f}") cols[1].metric("Perdita massima", f"${curve.max_loss_usd:+.2f}") cols[2].metric( "Breakeven", f"{curve.breakeven:.2f}" if curve.breakeven is not None else "—", ) def _render_decisions(position: PositionRecord) -> None: st.subheader("Storico decisioni") decisions = load_decisions_for_position(position.proposal_id) if not decisions: st.info("Nessuna decisione registrata per questa posizione.") return rows = [] for d in decisions: try: outputs = json.loads(d.outputs_json) except (TypeError, ValueError): outputs = {} rows.append( { "timestamp": humanize_dt(d.timestamp), "tipo decisione": d.decision_type, "azione": d.action_taken or "—", "note": d.notes or "", "output": json.dumps(outputs, sort_keys=True) if outputs else "", } ) st.dataframe(rows, use_container_width=True, hide_index=True) def render() -> None: st.title("💼 Posizione") st.caption( "Drilldown sul trade: gambe, payoff a scadenza, storico decisioni. " "Tutti i dati arrivano da SQLite — nessuna chiamata MCP live." ) db_path = _resolve_db() open_pos = load_open_positions(db_path=db_path) closed_recent = load_closed_positions(db_path=db_path)[-10:] candidates: list[PositionRecord] = list(open_pos) + list(reversed(closed_recent)) if not candidates: st.info( "Nessuna posizione da mostrare. La pagina si popolerà non " "appena il motore aprirà il primo trade." ) return labels = {_position_label(p): p for p in candidates} pick = st.selectbox( "Posizione", options=list(labels.keys()), index=0, ) position = labels[pick] # Deep-link via ?proposal_id=… qp = st.query_params.get("proposal_id") if qp: try: qp_uuid = UUID(qp) override = load_position_by_id(qp_uuid, db_path=db_path) if override is not None: position = override except ValueError: st.warning(f"Parametro proposal_id non valido: {qp}") st.divider() _render_header(position) st.divider() _render_distance(position) st.divider() _render_legs(position) st.divider() _render_payoff(position) st.divider() _render_decisions(position) render()