from __future__ import annotations import time import pandas as pd import streamlit as st from web_core.analytics import backtest_signals, simulate_trend_trades from web_core.chart_overlays import add_example_trade_markers, highlight_selected_trade from web_core.charting import build_figure from web_core.constants import TREND_BEAR, TREND_BULL, TREND_NEUTRAL from web_core.data import fetch_ohlc, maybe_drop_live_bar from web_core.exporting import df_for_export from web_core.insights import ( compare_symbols, compute_session_stats, compute_signal_quality, detect_regime, multi_timeframe_confirmation, send_webhook_alert, ) from web_core.ui.login_ui import render_profile_login from web_core.live_guide import build_live_decision_guide from web_core.auth.profile_store import ( find_existing_profile_id, first_query_param_value, is_truthy_flag, is_profile_session_expired, list_web_profiles, mark_profile_login, normalize_profile_id, normalize_web_settings, profile_requires_pin, save_web_settings, ) from web_core.ui.sidebar_ui import render_sidebar from web_core.strategy import classify_bars, detect_trends from web_core.market.symbols import resolve_symbol_identity from web_core.ui.training_ui import render_beginner_training_panel, render_training_panel from web_core.time_display import format_timestamp def main() -> None: st.set_page_config(page_title="Real Bars vs Fake Bars Analyzer", layout="wide") st.title("Real Bars vs Fake Bars Trend Analyzer") st.caption( "Price-action tool that classifies closed bars, filters fake bars, and tracks trend persistence using only real bars." ) st.caption("If you refresh, a brief placeholder screen may appear while data and UI load.") query_params = st.query_params now_epoch = time.time() active_profile = st.session_state.get("active_profile") last_active = st.session_state.get("profile_last_active_at") remember_requested = is_truthy_flag(first_query_param_value(query_params, "remember")) session_expired = False if active_profile and is_profile_session_expired(last_active, now_epoch): if remember_requested and not profile_requires_pin(str(active_profile)): st.session_state["profile_last_active_at"] = now_epoch else: session_expired = True st.session_state.pop("active_profile", None) st.session_state.pop("profile_last_active_at", None) active_profile = None if "profile" in query_params: del query_params["profile"] if not active_profile: remembered_profile = first_query_param_value(query_params, "profile") if remember_requested and remembered_profile: matched_profile = find_existing_profile_id(remembered_profile, set(list_web_profiles())) if matched_profile and not profile_requires_pin(matched_profile): active_profile = matched_profile st.session_state["active_profile"] = matched_profile st.session_state["profile_last_active_at"] = now_epoch mark_profile_login(matched_profile, now_epoch=int(now_epoch)) if not active_profile: render_profile_login(now_epoch=now_epoch, query_params=query_params, session_expired=session_expired) st.stop() active_profile = normalize_profile_id(active_profile) previous_session_profile = st.session_state.get("session_profile_id") if previous_session_profile != active_profile or "session_started_at" not in st.session_state: st.session_state["session_started_at"] = now_epoch st.session_state["session_profile_id"] = active_profile st.session_state["active_profile"] = active_profile st.session_state["profile_last_active_at"] = now_epoch sidebar_settings = normalize_web_settings( render_sidebar(active_profile=active_profile, query_params=query_params) ) try: save_web_settings(sidebar_settings, profile_id=active_profile) except Exception: pass symbol = str(sidebar_settings["symbol"]) interval = str(sidebar_settings["interval"]) period = str(sidebar_settings["period"]) max_bars = int(sidebar_settings["max_bars"]) display_timezone = str(sidebar_settings.get("display_timezone", "America/Chicago")) use_24h_time = bool(sidebar_settings.get("use_24h_time", False)) if not symbol: st.error("Please enter a symbol.") st.stop() symbol_identity = resolve_symbol_identity(symbol) identity_name = symbol_identity["name"] identity_exchange = symbol_identity["exchange"] if identity_name: st.markdown(f"### {symbol} - {identity_name}") if identity_exchange: st.caption(f"Exchange: {identity_exchange}") else: st.markdown(f"### {symbol}") try: raw = fetch_ohlc(symbol=symbol, interval=interval, period=period) raw = maybe_drop_live_bar(raw, interval=interval, enabled=bool(sidebar_settings["drop_live"])) if len(raw) > max_bars: raw = raw.iloc[-max_bars:].copy() if len(raw) < 3: st.error("Not enough bars to classify. Increase period or use a broader timeframe.") st.stop() classified = classify_bars( raw, use_body_range=bool(sidebar_settings["use_body_range"]), volume_filter_enabled=bool(sidebar_settings["volume_filter_enabled"]), volume_sma_window=int(sidebar_settings["volume_sma_window"]), volume_multiplier=float(sidebar_settings["volume_multiplier"]), ) analyzed, events = detect_trends(classified) except Exception as exc: st.error(f"Data error: {exc}") st.stop() replay_mode_enabled = bool(sidebar_settings.get("replay_mode_enabled", False)) replay_bars = int(sidebar_settings.get("replay_bars", 120)) if replay_mode_enabled: replay_len = max(3, min(len(analyzed), replay_bars)) analyzed_view = analyzed.iloc[:replay_len].copy() events_view = [e for e in events if e.timestamp <= analyzed_view.index[-1]] st.caption(f"Replay mode active: showing first {replay_len} bars only.") else: analyzed_view = analyzed events_view = events regime_label = detect_regime(analyzed_view) quality = compute_signal_quality( analyzed_view, trend_now=str(analyzed_view.iloc[-1]["trend_state"]), volume_filter_enabled=bool(sidebar_settings["volume_filter_enabled"]), ) if bool(sidebar_settings["enable_regime_filter"]) and regime_label == "Choppy": st.warning("Regime filter active: market is choppy. Directional bias is suppressed.") latest = analyzed_view.iloc[-1] trend_now = str(latest["trend_state"]) if bool(sidebar_settings["enable_regime_filter"]) and regime_label == "Choppy": trend_now = TREND_NEUTRAL c1, c2, c3, c4 = st.columns(4) c1.metric("Current Trend", trend_now) c2.metric("Real Bullish Bars", int((analyzed_view["classification"] == "real_bull").sum())) c3.metric("Real Bearish Bars", int((analyzed_view["classification"] == "real_bear").sum())) c4.metric("Fake Bars", int((analyzed_view["classification"] == "fake").sum())) previous_trend = str(analyzed_view.iloc[-2]["trend_state"]) if len(analyzed_view) > 1 else TREND_NEUTRAL latest_classification = str(latest["classification"]) live_guide = build_live_decision_guide(trend_now, previous_trend, latest_classification) q1, q2, q3 = st.columns(3) q1.metric("Signal Quality", f"{quality['score']} ({quality['label']})") q2.metric("Regime", regime_label) q3.metric("Recent Fake Ratio", f"{quality['fake_ratio']}%") render_beginner_training_panel( analyzed=analyzed_view, trend_now=trend_now, signal_quality=quality, regime_label=regime_label, ) run_advanced_panels = bool(sidebar_settings.get("advanced_auto_run", False) or sidebar_settings.get("run_advanced_now", False)) if (bool(sidebar_settings["enable_multi_tf_confirmation"]) or bool(sidebar_settings["enable_compare_symbols"])) and not run_advanced_panels: st.caption("Advanced panels paused to keep UI responsive. Click 'Run advanced panels now' in sidebar.") if bool(sidebar_settings["enable_multi_tf_confirmation"]) and run_advanced_panels: mtf = multi_timeframe_confirmation( symbol=symbol, period=period, use_body_range=bool(sidebar_settings["use_body_range"]), volume_filter_enabled=bool(sidebar_settings["volume_filter_enabled"]), volume_sma_window=int(sidebar_settings["volume_sma_window"]), volume_multiplier=float(sidebar_settings["volume_multiplier"]), ) st.subheader("Multi-Timeframe Confirmation") st.caption(f"Consensus: {mtf['consensus']} | Agreement: {mtf['agreement_pct']}%") st.dataframe(pd.DataFrame(mtf["states"]), use_container_width=True) if bool(sidebar_settings["show_live_guide"]): st.subheader("Live Decision Guide") g1, g2, g3 = st.columns(3) g1.metric("Bias", live_guide["bias"]) g2.metric("Signal Status", live_guide["confirmation"]) g3.metric("Latest Bar", latest_classification) st.caption(live_guide["confirmation_detail"]) st.caption(live_guide["classification_hint"]) st.info(live_guide["action"]) st.caption(f"Invalidation rule: {live_guide['invalidation']}") alert_key = f"{active_profile}-{symbol}-{interval}-{period}" newest_event = events_view[-1].event if events_view else "" previous_event = st.session_state.get(f"last_event-{alert_key}", "") if newest_event and newest_event != previous_event: st.warning(f"Alert: {newest_event}") st.session_state[f"last_event-{alert_key}"] = newest_event if bool(sidebar_settings["enable_alert_rules"]): should_alert = ( (bool(sidebar_settings["alert_on_bull"]) and "Bullish" in newest_event) or (bool(sidebar_settings["alert_on_bear"]) and "Bearish" in newest_event) ) if should_alert: payload = { "symbol": symbol, "interval": interval, "event": newest_event, "trend_now": trend_now, "timestamp": str(analyzed_view.index[-1]), } ok, detail = send_webhook_alert(str(sidebar_settings["alert_webhook_url"]), payload) if str(sidebar_settings["alert_webhook_url"]).strip(): st.caption(f"Webhook alert status: {'sent' if ok else 'failed'} ({detail})") example_trades = simulate_trend_trades(analyzed_view, max_examples=int(sidebar_settings["max_training_examples"])) selected_trade = render_training_panel( show_past_behavior=bool(sidebar_settings["show_past_behavior"]), example_trades=example_trades, alert_key=alert_key, display_timezone=display_timezone, use_24h_time=use_24h_time, ) fig = build_figure( analyzed_view, gray_fake=bool(sidebar_settings["gray_fake"]), interval=interval, hide_market_closed_gaps=bool(sidebar_settings["hide_market_closed_gaps"]), display_timezone=display_timezone, use_24h_time=use_24h_time, ) if bool(sidebar_settings["show_trade_markers"]): add_example_trade_markers(fig, example_trades) highlight_selected_trade( fig, analyzed_view, selected_trade, focus_chart_on_selected_example=bool(sidebar_settings["focus_chart_on_selected_example"]), ) st.plotly_chart(fig, use_container_width=True) try: bt = backtest_signals( analyzed_view, slippage_bps=float(sidebar_settings["backtest_slippage_bps"]), fee_bps=float(sidebar_settings["backtest_fee_bps"]), stop_loss_pct=( float(sidebar_settings["backtest_stop_loss_pct"]) if float(sidebar_settings["backtest_stop_loss_pct"]) > 0 else None ), take_profit_pct=( float(sidebar_settings["backtest_take_profit_pct"]) if float(sidebar_settings["backtest_take_profit_pct"]) > 0 else None ), min_hold_bars=int(sidebar_settings["backtest_min_hold_bars"]), max_hold_bars=int(sidebar_settings["backtest_max_hold_bars"]), ) except Exception as exc: st.error(f"Backtest error: {exc}") bt = {"trades": 0, "wins": 0, "losses": 0, "win_rate": 0.0, "avg_pnl_pct": 0.0} st.subheader("Backtest Snapshot") b1, b2, b3, b4, b5 = st.columns(5) b1.metric("Signals", int(bt["trades"])) b2.metric("Wins", int(bt["wins"])) b3.metric("Losses", int(bt["losses"])) b4.metric("Win Rate", f"{bt['win_rate']}%") b5.metric("Avg P/L", f"{bt['avg_pnl_pct']}%") st.caption("Method: trend-change signal, scored by next-bar direction. Educational only; not a trading recommendation.") st.subheader("Session Stats") session_stats = compute_session_stats(analyzed_view, session_started_at=float(st.session_state["session_started_at"])) s1, s2, s3, s4 = st.columns(4) s1.metric("Since Login Wins", int(session_stats["wins"])) s2.metric("Since Login Losses", int(session_stats["losses"])) s3.metric("Since Login Avg Move", f"{session_stats['avg_move_pct']}%") s4.metric("Since Login Fake Rate", f"{session_stats['fake_ratio_pct']}%") if bool(sidebar_settings["enable_compare_symbols"]) and sidebar_settings["compare_symbols"] and run_advanced_panels: st.subheader("Compare Symbols") compare_df = compare_symbols( symbols=list(sidebar_settings["compare_symbols"]), interval=interval, period=period, use_body_range=bool(sidebar_settings["use_body_range"]), volume_filter_enabled=bool(sidebar_settings["volume_filter_enabled"]), volume_sma_window=int(sidebar_settings["volume_sma_window"]), volume_multiplier=float(sidebar_settings["volume_multiplier"]), ) st.dataframe(compare_df, use_container_width=True) st.subheader("Trend Events") if events_view: event_df = pd.DataFrame( { "timestamp": [ format_timestamp(e.timestamp, display_timezone=display_timezone, use_24h_time=use_24h_time) for e in events_view[-25:] ][::-1], "event": [e.event for e in events_view[-25:]][::-1], "trend_after": [e.trend_after for e in events_view[-25:]][::-1], } ) st.dataframe(event_df, use_container_width=True) else: st.info("No trend start/reversal events detected in the selected data window.") st.subheader("Export") export_df = df_for_export(analyzed_view) csv_bytes = export_df.to_csv(index=False).encode("utf-8") st.download_button( "Download classified data (CSV)", data=csv_bytes, file_name=f"{symbol}_{interval}_classified.csv", mime="text/csv", ) try: pdf_bytes = fig.to_image(format="pdf") st.download_button( "Download chart (PDF)", data=pdf_bytes, file_name=f"{symbol}_{interval}_chart.pdf", mime="application/pdf", ) except Exception: st.caption("PDF export unavailable. Install `kaleido` and rerun to enable chart PDF downloads.") with st.expander("Latest classified bars"): st.dataframe(export_df.tail(30), use_container_width=True) if __name__ == "__main__": main()