Signed-off-by: Matt Bruce <mbrucedogs@gmail.com>

This commit is contained in:
Matt Bruce 2026-02-16 22:53:58 -06:00
parent 1e73d49aa1
commit f731aa90e5
10 changed files with 813 additions and 87 deletions

View File

@ -96,6 +96,24 @@ Tip: sign and notarize before sharing broadly, so macOS trust prompts are reduce
14. Use `Live Decision Guide` to translate trend state into a practical bias/action/invalidation workflow. 14. Use `Live Decision Guide` to translate trend state into a practical bias/action/invalidation workflow.
15. Keep `Show past behavior examples` ON while learning to review historical entry/exit outcomes. 15. Keep `Show past behavior examples` ON while learning to review historical entry/exit outcomes.
## 4.1) Advanced Features (Optional)
- `Advanced Signals`:
- `Auto-run advanced panels (slower)`: keeps multi-timeframe/compare updating every rerun
- `Run advanced panels now`: one-shot run to avoid UI hangs during normal editing
- `Show multi-timeframe confirmation (1h/4h/1d)`
- `Regime filter (stand aside in choppy periods)`
- `Training & Guidance`:
- `Replay mode (hide future bars)` + `Replay bars shown`
- `Compare Symbols`:
- enable panel and provide comma-separated symbols (`AAPL, MSFT, NVDA`)
- `Alerts`:
- enable bullish/bearish rule alerts
- optional webhook URL for external delivery (Telegram/email automation via Zapier/Make/webhook adapters)
- `Backtest Controls`:
- slippage/fee bps
- stop-loss/take-profit %
- min/max hold bars
## 5) How To Read The Chart ## 5) How To Read The Chart
- Candle layer: full price action - Candle layer: full price action
- Green triangle-up markers: `real_bull` - Green triangle-up markers: `real_bull`
@ -224,6 +242,10 @@ If PDF fails, ensure `kaleido` is installed (already in `requirements.txt`).
./run.sh ./run.sh
``` ```
### I see a blank/skeleton placeholder on refresh
- A brief placeholder can appear while Streamlit initializes the page.
- Wait a moment; this is expected loading behavior, not an app crash.
### Port already in use ### Port already in use
```bash ```bash
streamlit run app.py --server.port 8502 streamlit run app.py --server.port 8502
@ -241,7 +263,16 @@ streamlit run app.py --server.port 8502
### Watchlist or preset not applying as expected ### Watchlist or preset not applying as expected
- Watchlist uses comma-separated symbols and deduplicates automatically. - Watchlist uses comma-separated symbols and deduplicates automatically.
- Preset `Custom` disables forced defaults; other presets re-apply tuned defaults each run. - Select a non-`Custom` preset and click `Apply Preset` to populate defaults.
- After applying, you can still manually tweak any field.
### Replay mode looks different than normal view
- Replay mode intentionally truncates visible bars to simulate no-hindsight training.
- Disable replay mode to return to full-history analysis.
### Alert webhook errors
- Confirm webhook URL accepts JSON `POST`.
- For Telegram/email, route through an automation endpoint that supports incoming webhooks.
### PIN login fails ### PIN login fails
- Ensure the profile name is correct (name matching is case-insensitive). - Ensure the profile name is correct (name matching is case-insensitive).

View File

@ -23,6 +23,7 @@ Provide an analysis-only charting tool that classifies OHLC bars as real/fake, t
- `max_bars` - `max_bars`
- optional `market_preset` - optional `market_preset`
- filter/toggle settings - filter/toggle settings
- optional advanced controls (alerts, replay, compare symbols, backtest controls, regime filter)
2. App fetches OHLCV via Yahoo Finance (`yfinance`). 2. App fetches OHLCV via Yahoo Finance (`yfinance`).
3. Optional last-bar drop (live-bar guard) for intraday intervals. 3. Optional last-bar drop (live-bar guard) for intraday intervals.
4. Bars are classified (`real_bull`, `real_bear`, `fake`, `unclassified` for first bar). 4. Bars are classified (`real_bull`, `real_bear`, `fake`, `unclassified` for first bar).
@ -67,6 +68,16 @@ Watchlist and preset behavior:
- `market_preset` applies tuned defaults for common workflows (`Custom`, `Stocks Swing`, `Crypto Intraday`, `Crypto Swing`). - `market_preset` applies tuned defaults for common workflows (`Custom`, `Stocks Swing`, `Crypto Intraday`, `Crypto Swing`).
- Preset application affects defaults for timeframe/period/max bars and related monitoring/gap options. - Preset application affects defaults for timeframe/period/max bars and related monitoring/gap options.
Advanced feature behavior:
- Multi-timeframe confirmation (`1h`, `4h`, `1d`) can show consensus trend and agreement percent.
- Alert rules can trigger on bullish/bearish trend events and optionally POST JSON payloads to a webhook URL.
- Session stats track since-login wins/losses, average move after signal, and fake-bar rate for visible data.
- Backtest controls include slippage/fees (bps), optional stop-loss/take-profit %, and min/max hold bars.
- Signal quality score combines trend state, volume-filter usage, and recent fake-bar density.
- Replay mode limits visible bars to first N rows to simulate no-hindsight review.
- Compare symbols panel can summarize trend/regime/fake-ratio for a small basket.
- Regime filter flags choppy periods and can suppress directional bias.
Normalization constraints: Normalization constraints:
- `symbol`: uppercase, non-empty fallback `AAPL` - `symbol`: uppercase, non-empty fallback `AAPL`
- `interval`: must be one of `INTERVAL_OPTIONS`, fallback `1d` - `interval`: must be one of `INTERVAL_OPTIONS`, fallback `1d`
@ -82,6 +93,14 @@ Normalization constraints:
- `max_training_examples`: `[5, 100]`, fallback `20` - `max_training_examples`: `[5, 100]`, fallback `20`
- `watchlist`: uppercase de-duplicated symbol list, max 40 items - `watchlist`: uppercase de-duplicated symbol list, max 40 items
- `market_preset`: one of `Custom`, `Stocks Swing`, `Crypto Intraday`, `Crypto Swing`, fallback `Custom` - `market_preset`: one of `Custom`, `Stocks Swing`, `Crypto Intraday`, `Crypto Swing`, fallback `Custom`
- `compare_symbols`: uppercase de-duplicated symbol list, max 12 items
- `alert_webhook_url`: optional string
- `backtest_slippage_bps`: `[0.0, 100.0]`, fallback `0.0`
- `backtest_fee_bps`: `[0.0, 100.0]`, fallback `0.0`
- `backtest_stop_loss_pct`: `[0.0, 25.0]`, fallback `0.0` (0 disables)
- `backtest_take_profit_pct`: `[0.0, 25.0]`, fallback `0.0` (0 disables)
- `backtest_min_hold_bars`: `[1, 20]`, fallback `1`
- `backtest_max_hold_bars`: `[1, 40]`, fallback `1` and clamped to `>= min_hold`
- booleans normalized from common truthy/falsy strings and numbers - booleans normalized from common truthy/falsy strings and numbers
## 5. Classification Rules ## 5. Classification Rules
@ -162,7 +181,7 @@ Gap handling (`hide_market_closed_gaps`):
- Trend events table (latest events) - Trend events table (latest events)
- Backtest snapshot: - Backtest snapshot:
- signal at trend-change rows to active bull/bear states - signal at trend-change rows to active bull/bear states
- next-bar close determines win/loss - advanced mode supports configurable costs/holds/stop/target
- Past behavior examples (optional training panel): - Past behavior examples (optional training panel):
- historical examples using trend-confirmation entries and opposite-confirmation exits - historical examples using trend-confirmation entries and opposite-confirmation exits
- per-example direction, entry/exit timestamps, bars held, P/L%, and outcome - per-example direction, entry/exit timestamps, bars held, P/L%, and outcome
@ -172,6 +191,10 @@ Gap handling (`hide_market_closed_gaps`):
- Exports: - Exports:
- CSV always available - CSV always available
- PDF via Plotly image export (requires Kaleido runtime) - PDF via Plotly image export (requires Kaleido runtime)
- Additional optional panels:
- multi-timeframe confirmation
- compare symbols snapshot
- session stats
## 10. Validation Expectations ## 10. Validation Expectations
Code-level checks: Code-level checks:

View File

@ -8,9 +8,17 @@ import streamlit as st
from web_core.analytics import backtest_signals, simulate_trend_trades 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.chart_overlays import add_example_trade_markers, highlight_selected_trade
from web_core.charting import build_figure from web_core.charting import build_figure
from web_core.constants import TREND_NEUTRAL 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.data import fetch_ohlc, maybe_drop_live_bar
from web_core.exporting import df_for_export 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.ui.login_ui import render_profile_login
from web_core.live_guide import build_live_decision_guide from web_core.live_guide import build_live_decision_guide
from web_core.auth.profile_store import ( from web_core.auth.profile_store import (
@ -21,6 +29,7 @@ from web_core.auth.profile_store import (
list_web_profiles, list_web_profiles,
mark_profile_login, mark_profile_login,
normalize_profile_id, normalize_profile_id,
normalize_web_settings,
profile_requires_pin, profile_requires_pin,
save_web_settings, save_web_settings,
) )
@ -36,6 +45,7 @@ def main() -> None:
st.caption( st.caption(
"Price-action tool that classifies closed bars, filters fake bars, and tracks trend persistence using only real bars." "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 query_params = st.query_params
now_epoch = time.time() now_epoch = time.time()
@ -70,10 +80,16 @@ def main() -> None:
st.stop() st.stop()
active_profile = normalize_profile_id(active_profile) 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["active_profile"] = active_profile
st.session_state["profile_last_active_at"] = now_epoch st.session_state["profile_last_active_at"] = now_epoch
sidebar_settings = render_sidebar(active_profile=active_profile, query_params=query_params) sidebar_settings = normalize_web_settings(
render_sidebar(active_profile=active_profile, query_params=query_params)
)
try: try:
save_web_settings(sidebar_settings, profile_id=active_profile) save_web_settings(sidebar_settings, profile_id=active_profile)
@ -121,19 +137,64 @@ def main() -> None:
st.error(f"Data error: {exc}") st.error(f"Data error: {exc}")
st.stop() st.stop()
latest = analyzed.iloc[-1] 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"]) 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, c2, c3, c4 = st.columns(4)
c1.metric("Current Trend", trend_now) c1.metric("Current Trend", trend_now)
c2.metric("Real Bullish Bars", int((analyzed["classification"] == "real_bull").sum())) c2.metric("Real Bullish Bars", int((analyzed_view["classification"] == "real_bull").sum()))
c3.metric("Real Bearish Bars", int((analyzed["classification"] == "real_bear").sum())) c3.metric("Real Bearish Bars", int((analyzed_view["classification"] == "real_bear").sum()))
c4.metric("Fake Bars", int((analyzed["classification"] == "fake").sum())) c4.metric("Fake Bars", int((analyzed_view["classification"] == "fake").sum()))
previous_trend = str(analyzed.iloc[-2]["trend_state"]) if len(analyzed) > 1 else TREND_NEUTRAL previous_trend = str(analyzed_view.iloc[-2]["trend_state"]) if len(analyzed_view) > 1 else TREND_NEUTRAL
latest_classification = str(latest["classification"]) latest_classification = str(latest["classification"])
live_guide = build_live_decision_guide(trend_now, previous_trend, 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']}%")
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"]): if bool(sidebar_settings["show_live_guide"]):
st.subheader("Live Decision Guide") st.subheader("Live Decision Guide")
g1, g2, g3 = st.columns(3) g1, g2, g3 = st.columns(3)
@ -146,13 +207,29 @@ def main() -> None:
st.caption(f"Invalidation rule: {live_guide['invalidation']}") st.caption(f"Invalidation rule: {live_guide['invalidation']}")
alert_key = f"{active_profile}-{symbol}-{interval}-{period}" alert_key = f"{active_profile}-{symbol}-{interval}-{period}"
newest_event = events[-1].event if events else "" newest_event = events_view[-1].event if events_view else ""
previous_event = st.session_state.get(f"last_event-{alert_key}", "") previous_event = st.session_state.get(f"last_event-{alert_key}", "")
if newest_event and newest_event != previous_event: if newest_event and newest_event != previous_event:
st.warning(f"Alert: {newest_event}") st.warning(f"Alert: {newest_event}")
st.session_state[f"last_event-{alert_key}"] = 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, max_examples=int(sidebar_settings["max_training_examples"])) example_trades = simulate_trend_trades(analyzed_view, max_examples=int(sidebar_settings["max_training_examples"]))
selected_trade = render_training_panel( selected_trade = render_training_panel(
show_past_behavior=bool(sidebar_settings["show_past_behavior"]), show_past_behavior=bool(sidebar_settings["show_past_behavior"]),
example_trades=example_trades, example_trades=example_trades,
@ -160,7 +237,7 @@ def main() -> None:
) )
fig = build_figure( fig = build_figure(
analyzed, analyzed_view,
gray_fake=bool(sidebar_settings["gray_fake"]), gray_fake=bool(sidebar_settings["gray_fake"]),
interval=interval, interval=interval,
hide_market_closed_gaps=bool(sidebar_settings["hide_market_closed_gaps"]), hide_market_closed_gaps=bool(sidebar_settings["hide_market_closed_gaps"]),
@ -169,28 +246,70 @@ def main() -> None:
add_example_trade_markers(fig, example_trades) add_example_trade_markers(fig, example_trades)
highlight_selected_trade( highlight_selected_trade(
fig, fig,
analyzed, analyzed_view,
selected_trade, selected_trade,
focus_chart_on_selected_example=bool(sidebar_settings["focus_chart_on_selected_example"]), focus_chart_on_selected_example=bool(sidebar_settings["focus_chart_on_selected_example"]),
) )
st.plotly_chart(fig, use_container_width=True) st.plotly_chart(fig, use_container_width=True)
bt = backtest_signals(analyzed) 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") st.subheader("Backtest Snapshot")
b1, b2, b3, b4 = st.columns(4) b1, b2, b3, b4, b5 = st.columns(5)
b1.metric("Signals", int(bt["trades"])) b1.metric("Signals", int(bt["trades"]))
b2.metric("Wins", int(bt["wins"])) b2.metric("Wins", int(bt["wins"]))
b3.metric("Losses", int(bt["losses"])) b3.metric("Losses", int(bt["losses"]))
b4.metric("Win Rate", f"{bt['win_rate']}%") 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.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") st.subheader("Trend Events")
if events: if events_view:
event_df = pd.DataFrame( event_df = pd.DataFrame(
{ {
"timestamp": [str(e.timestamp) for e in events[-25:]][::-1], "timestamp": [str(e.timestamp) for e in events_view[-25:]][::-1],
"event": [e.event for e in events[-25:]][::-1], "event": [e.event for e in events_view[-25:]][::-1],
"trend_after": [e.trend_after for e in events[-25:]][::-1], "trend_after": [e.trend_after for e in events_view[-25:]][::-1],
} }
) )
st.dataframe(event_df, use_container_width=True) st.dataframe(event_df, use_container_width=True)
@ -198,7 +317,7 @@ def main() -> None:
st.info("No trend start/reversal events detected in the selected data window.") st.info("No trend start/reversal events detected in the selected data window.")
st.subheader("Export") st.subheader("Export")
export_df = df_for_export(analyzed) export_df = df_for_export(analyzed_view)
csv_bytes = export_df.to_csv(index=False).encode("utf-8") csv_bytes = export_df.to_csv(index=False).encode("utf-8")
st.download_button( st.download_button(
"Download classified data (CSV)", "Download classified data (CSV)",

View File

@ -0,0 +1,39 @@
from __future__ import annotations
import pandas as pd
from web_core.analytics import backtest_signals
from web_core.constants import TREND_BEAR, TREND_BULL, TREND_NEUTRAL
def test_backtest_signals_supports_cost_and_hold_controls() -> None:
idx = pd.date_range("2025-01-01", periods=8, freq="D")
df = pd.DataFrame(
{
"Close": [100.0, 101.0, 103.0, 104.0, 102.0, 99.0, 97.0, 98.0],
"trend_state": [
TREND_NEUTRAL,
TREND_BULL,
TREND_BULL,
TREND_BEAR,
TREND_BEAR,
TREND_BULL,
TREND_BULL,
TREND_BULL,
],
},
index=idx,
)
out = backtest_signals(
df,
slippage_bps=5.0,
fee_bps=2.0,
stop_loss_pct=2.0,
take_profit_pct=3.0,
min_hold_bars=1,
max_hold_bars=3,
)
assert out["trades"] >= 1
assert "avg_pnl_pct" in out

View File

@ -0,0 +1,36 @@
from __future__ import annotations
import pandas as pd
from web_core.constants import TREND_BEAR, TREND_BULL, TREND_NEUTRAL
from web_core.insights import compute_session_stats, compute_signal_quality, detect_regime
def _make_analyzed(rows: int = 12) -> pd.DataFrame:
idx = pd.date_range("2025-01-01", periods=rows, freq="h")
classifications = ["fake", "real_bull", "fake", "real_bear"] * (rows // 4) + ["fake"] * (rows % 4)
trends = [TREND_NEUTRAL, TREND_BULL, TREND_BULL, TREND_BEAR] * (rows // 4) + [TREND_BEAR] * (rows % 4)
closes = [100 + i for i in range(rows)]
return pd.DataFrame({"classification": classifications[:rows], "trend_state": trends[:rows], "Close": closes}, index=idx)
def test_detect_regime_flags_choppy_when_fake_ratio_high() -> None:
df = _make_analyzed(12)
out = detect_regime(df, lookback=12)
assert out in {"Choppy", "Trending"}
def test_compute_signal_quality_returns_expected_keys() -> None:
df = _make_analyzed(20)
quality = compute_signal_quality(df, trend_now=TREND_BULL, volume_filter_enabled=True)
assert set(quality.keys()) == {"score", "label", "fake_ratio"}
assert quality["label"] in {"Low", "Medium", "High"}
def test_compute_session_stats_has_numeric_outputs() -> None:
df = _make_analyzed(20)
stats = compute_session_stats(df, session_started_at=0.0)
assert isinstance(stats["wins"], int)
assert isinstance(stats["losses"], int)
assert isinstance(stats["avg_move_pct"], float)
assert isinstance(stats["fake_ratio_pct"], float)

View File

@ -5,9 +5,23 @@ import pandas as pd
from .constants import TREND_BEAR, TREND_BULL from .constants import TREND_BEAR, TREND_BULL
def backtest_signals(df: pd.DataFrame) -> dict[str, float | int]: def _trade_return_pct(direction: str, entry_price: float, exit_price: float) -> float:
if direction == "LONG":
return ((exit_price - entry_price) / entry_price) * 100.0
return ((entry_price - exit_price) / entry_price) * 100.0
def backtest_signals(
df: pd.DataFrame,
slippage_bps: float = 0.0,
fee_bps: float = 0.0,
stop_loss_pct: float | None = None,
take_profit_pct: float | None = None,
max_hold_bars: int = 1,
min_hold_bars: int = 1,
) -> dict[str, float | int]:
if len(df) < 4: if len(df) < 4:
return {"trades": 0, "wins": 0, "losses": 0, "win_rate": 0.0} return {"trades": 0, "wins": 0, "losses": 0, "win_rate": 0.0, "avg_pnl_pct": 0.0}
trend_series = df["trend_state"] trend_series = df["trend_state"]
trend_change = trend_series != trend_series.shift(1) trend_change = trend_series != trend_series.shift(1)
@ -15,6 +29,10 @@ def backtest_signals(df: pd.DataFrame) -> dict[str, float | int]:
wins = 0 wins = 0
losses = 0 losses = 0
pnl_values: list[float] = []
max_hold = max(1, int(max_hold_bars))
min_hold = max(1, min(int(min_hold_bars), max_hold))
total_cost_pct = ((float(slippage_bps) + float(fee_bps)) * 2.0) / 100.0
for idx in signal_idx: for idx in signal_idx:
pos = df.index.get_loc(idx) pos = df.index.get_loc(idx)
@ -22,19 +40,37 @@ def backtest_signals(df: pd.DataFrame) -> dict[str, float | int]:
continue continue
entry_close = float(df.iloc[pos]["Close"]) entry_close = float(df.iloc[pos]["Close"])
next_close = float(df.iloc[pos + 1]["Close"])
signal_trend = df.iloc[pos]["trend_state"] signal_trend = df.iloc[pos]["trend_state"]
direction = "LONG" if signal_trend == TREND_BULL else "SHORT"
if signal_trend == TREND_BULL: exit_price = float(df.iloc[min(pos + max_hold, len(df) - 1)]["Close"])
wins += int(next_close > entry_close) for hold in range(min_hold, max_hold + 1):
losses += int(next_close <= entry_close) exit_pos = min(pos + hold, len(df) - 1)
elif signal_trend == TREND_BEAR: candidate_exit = float(df.iloc[exit_pos]["Close"])
wins += int(next_close < entry_close) raw_pnl = _trade_return_pct(direction, entry_close, candidate_exit)
losses += int(next_close >= entry_close) if stop_loss_pct is not None and raw_pnl <= -float(stop_loss_pct):
exit_price = candidate_exit
break
if take_profit_pct is not None and raw_pnl >= float(take_profit_pct):
exit_price = candidate_exit
break
exit_price = candidate_exit
pnl_pct = _trade_return_pct(direction, entry_close, exit_price) - total_cost_pct
pnl_values.append(pnl_pct)
wins += int(pnl_pct > 0)
losses += int(pnl_pct <= 0)
trades = wins + losses trades = wins + losses
win_rate = (wins / trades * 100.0) if trades else 0.0 win_rate = (wins / trades * 100.0) if trades else 0.0
return {"trades": trades, "wins": wins, "losses": losses, "win_rate": round(win_rate, 2)} avg_pnl_pct = (sum(pnl_values) / len(pnl_values)) if pnl_values else 0.0
return {
"trades": trades,
"wins": wins,
"losses": losses,
"win_rate": round(win_rate, 2),
"avg_pnl_pct": round(avg_pnl_pct, 2),
}
def simulate_trend_trades(df: pd.DataFrame, max_examples: int = 20) -> pd.DataFrame: def simulate_trend_trades(df: pd.DataFrame, max_examples: int = 20) -> pd.DataFrame:

View File

@ -31,6 +31,41 @@
<h2>How To Use The New Training Features</h2> <h2>How To Use The New Training Features</h2>
<p>Training toggles are OFF by default.</p> <p>Training toggles are OFF by default.</p>
<h2>Advanced Panels</h2>
<p>These are in the sidebar as collapsible sections.</p>
<h3>Advanced Signals</h3>
<ul>
<li><code>Auto-run advanced panels (slower)</code>: keeps advanced calculations running on every rerun.</li>
<li><code>Run advanced panels now</code>: one-shot run when auto-run is OFF.</li>
<li><code>Show multi-timeframe confirmation (1h/4h/1d)</code>: shows consensus and agreement %.</li>
<li><code>Regime filter</code>: flags choppy conditions and can suppress directional bias.</li>
</ul>
<h3>Training &amp; Guidance (Replay)</h3>
<ul>
<li><code>Replay mode</code> + <code>Replay bars shown</code>: hides future bars for practice mode.</li>
</ul>
<h3>Compare Symbols</h3>
<ul>
<li>Enable panel and enter comma-separated symbols (example: <code>AAPL, MSFT, NVDA</code>).</li>
<li>Shows per-symbol trend state, regime, and fake-bar density snapshot.</li>
</ul>
<h3>Alerts</h3>
<ul>
<li>Enable bullish and/or bearish event alerts.</li>
<li>Optional webhook URL sends JSON alerts to external automations (Zapier/Make/Telegram bridge).</li>
</ul>
<h3>Backtest Controls</h3>
<ul>
<li><code>Slippage (bps per side)</code> and <code>Fee (bps per side)</code>: execution cost realism.</li>
<li><code>Stop loss (%)</code> and <code>Take profit (%)</code>: optional exits; set 0 to disable.</li>
<li><code>Min hold bars</code> and <code>Max hold bars</code>: holding window constraints.</li>
</ul>
<h3>1) Live Decision Guide (what to do now)</h3> <h3>1) Live Decision Guide (what to do now)</h3>
<ul> <ul>
<li><strong>Bias</strong>: Long, Short, or Stand Aside / Neutral.</li> <li><strong>Bias</strong>: Long, Short, or Stand Aside / Neutral.</li>
@ -75,6 +110,7 @@
<h2>Important Notes</h2> <h2>Important Notes</h2>
<ul> <ul>
<li>On browser refresh, a brief placeholder screen can appear while Streamlit loads the app shell.</li>
<li>Analysis and training only. No trade execution.</li> <li>Analysis and training only. No trade execution.</li>
<li>Educational examples are simplified and not financial advice.</li> <li>Educational examples are simplified and not financial advice.</li>
<li>Use with risk controls before any real-money decision.</li> <li>Use with risk controls before any real-money decision.</li>

View File

@ -0,0 +1,203 @@
from __future__ import annotations
import json
import urllib.request
from datetime import datetime, timezone
from typing import Any
import pandas as pd
import streamlit as st
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.strategy import classify_bars, detect_trends
def detect_regime(analyzed: pd.DataFrame, lookback: int = 30) -> str:
if analyzed.empty:
return "Unknown"
window = analyzed.tail(max(5, lookback))
fake_ratio = float((window["classification"] == "fake").mean())
trend_changes = int((window["trend_state"] != window["trend_state"].shift(1)).sum())
if fake_ratio >= 0.55 or trend_changes >= max(3, len(window) // 8):
return "Choppy"
return "Trending"
def compute_signal_quality(
analyzed: pd.DataFrame,
trend_now: str,
volume_filter_enabled: bool,
lookback: int = 20,
) -> dict[str, float | str]:
if analyzed.empty:
return {"score": 0.0, "label": "Low", "fake_ratio": 1.0}
window = analyzed.tail(max(5, lookback))
fake_ratio = float((window["classification"] == "fake").mean())
real_ratio = 1.0 - fake_ratio
trend_bonus = 0.2 if trend_now in {TREND_BULL, TREND_BEAR} else 0.0
volume_bonus = 0.1 if volume_filter_enabled else 0.0
score = min(1.0, max(0.0, real_ratio * 0.7 + trend_bonus + volume_bonus))
if score >= 0.75:
label = "High"
elif score >= 0.5:
label = "Medium"
else:
label = "Low"
return {"score": round(score * 100.0, 1), "label": label, "fake_ratio": round(fake_ratio * 100.0, 1)}
@st.cache_data(show_spinner=False, ttl=120)
def analyze_symbol_state(
symbol: str,
interval: str,
period: str,
use_body_range: bool,
volume_filter_enabled: bool,
volume_sma_window: int,
volume_multiplier: float,
) -> dict[str, Any]:
try:
raw = fetch_ohlc(symbol=symbol, interval=interval, period=period)
raw = maybe_drop_live_bar(raw, interval=interval, enabled=True)
if len(raw) < 3:
return {"symbol": symbol, "trend": "Insufficient", "regime": "Unknown"}
analyzed, _ = detect_trends(
classify_bars(
raw,
use_body_range=use_body_range,
volume_filter_enabled=volume_filter_enabled,
volume_sma_window=volume_sma_window,
volume_multiplier=volume_multiplier,
)
)
trend_now = str(analyzed.iloc[-1]["trend_state"])
return {
"symbol": symbol,
"trend": trend_now,
"regime": detect_regime(analyzed),
"fake_ratio": round(float((analyzed.tail(20)["classification"] == "fake").mean()) * 100.0, 1),
}
except Exception as exc:
return {"symbol": symbol, "trend": "Error", "regime": "Unknown", "error": str(exc)}
def multi_timeframe_confirmation(
symbol: str,
period: str,
use_body_range: bool,
volume_filter_enabled: bool,
volume_sma_window: int,
volume_multiplier: float,
) -> dict[str, Any]:
intervals = ["1h", "4h", "1d"]
states = [
analyze_symbol_state(
symbol=symbol,
interval=itv,
period=period,
use_body_range=use_body_range,
volume_filter_enabled=volume_filter_enabled,
volume_sma_window=volume_sma_window,
volume_multiplier=volume_multiplier,
)
for itv in intervals
]
trends = [str(item.get("trend")) for item in states if str(item.get("trend")) in {TREND_BULL, TREND_BEAR}]
bull_count = trends.count(TREND_BULL)
bear_count = trends.count(TREND_BEAR)
agreement = max(bull_count, bear_count) / 3.0
consensus = TREND_NEUTRAL
if bull_count > bear_count:
consensus = TREND_BULL
elif bear_count > bull_count:
consensus = TREND_BEAR
return {"states": states, "agreement_pct": round(agreement * 100.0, 1), "consensus": consensus}
def compare_symbols(
symbols: list[str],
interval: str,
period: str,
use_body_range: bool,
volume_filter_enabled: bool,
volume_sma_window: int,
volume_multiplier: float,
) -> pd.DataFrame:
rows: list[dict[str, Any]] = []
for symbol in symbols:
state = analyze_symbol_state(
symbol=symbol,
interval=interval,
period=period,
use_body_range=use_body_range,
volume_filter_enabled=volume_filter_enabled,
volume_sma_window=volume_sma_window,
volume_multiplier=volume_multiplier,
)
rows.append(state)
return pd.DataFrame(rows)
def compute_session_stats(
analyzed: pd.DataFrame,
session_started_at: float,
) -> dict[str, float | int]:
if analyzed.empty:
return {"wins": 0, "losses": 0, "avg_move_pct": 0.0, "fake_ratio_pct": 0.0}
idx = analyzed.index
if idx.tz is None:
cutoff = pd.Timestamp(datetime.fromtimestamp(session_started_at))
else:
cutoff = pd.Timestamp(datetime.fromtimestamp(session_started_at, tz=timezone.utc)).tz_convert(idx.tz)
window = analyzed[analyzed.index >= cutoff]
if window.empty:
window = analyzed.tail(30)
wins = 0
losses = 0
moves: list[float] = []
for i in range(1, len(window) - 1):
trend = str(window.iloc[i]["trend_state"])
prev = str(window.iloc[i - 1]["trend_state"])
if trend == prev or trend not in {TREND_BULL, TREND_BEAR}:
continue
entry = float(window.iloc[i]["Close"])
nxt = float(window.iloc[i + 1]["Close"])
move = ((nxt - entry) / entry) * 100.0 if trend == TREND_BULL else ((entry - nxt) / entry) * 100.0
moves.append(move)
wins += int(move > 0)
losses += int(move <= 0)
fake_ratio_pct = float((window["classification"] == "fake").mean()) * 100.0
avg_move_pct = (sum(moves) / len(moves)) if moves else 0.0
return {
"wins": wins,
"losses": losses,
"avg_move_pct": round(avg_move_pct, 2),
"fake_ratio_pct": round(fake_ratio_pct, 2),
}
def send_webhook_alert(url: str, payload: dict[str, Any]) -> tuple[bool, str]:
target = str(url or "").strip()
if not target:
return False, "No webhook URL configured."
try:
body = json.dumps(payload).encode("utf-8")
request = urllib.request.Request(
target,
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=5) as response:
code = int(response.status)
return (200 <= code < 300), f"Webhook responded with status {code}"
except Exception as exc:
return False, str(exc)

View File

@ -64,6 +64,26 @@ def _normalize_watchlist(raw: Any) -> list[str]:
return symbols[:40] return symbols[:40]
def _normalize_symbol_list(raw: Any, limit: int = 20) -> list[str]:
tokens: list[str] = []
if isinstance(raw, list):
tokens = [str(item) for item in raw]
else:
tokens = [str(raw or "")]
seen: set[str] = set()
out: list[str] = []
for token in tokens:
normalized_token = token.replace("\\n", "\n").replace("\\N", "\n").replace(",", "\n")
for split_token in normalized_token.splitlines():
symbol = split_token.strip().upper()
if not symbol or symbol in seen:
continue
seen.add(symbol)
out.append(symbol)
return out[:limit]
def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]: def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]:
raw = raw or {} raw = raw or {}
defaults: dict[str, Any] = { defaults: dict[str, Any] = {
@ -87,6 +107,23 @@ def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]:
"refresh_sec": 60, "refresh_sec": 60,
"watchlist": [], "watchlist": [],
"market_preset": "Custom", "market_preset": "Custom",
"compare_symbols": [],
"enable_compare_symbols": False,
"enable_multi_tf_confirmation": False,
"enable_regime_filter": False,
"advanced_auto_run": False,
"replay_mode_enabled": False,
"replay_bars": 120,
"enable_alert_rules": False,
"alert_on_bull": True,
"alert_on_bear": True,
"alert_webhook_url": "",
"backtest_slippage_bps": 0.0,
"backtest_fee_bps": 0.0,
"backtest_stop_loss_pct": 0.0,
"backtest_take_profit_pct": 0.0,
"backtest_min_hold_bars": 1,
"backtest_max_hold_bars": 1,
} }
symbol = str(raw.get("symbol", defaults["symbol"])).strip().upper() symbol = str(raw.get("symbol", defaults["symbol"])).strip().upper()
@ -104,6 +141,21 @@ def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]:
market_preset = str(raw.get("market_preset", defaults["market_preset"])) market_preset = str(raw.get("market_preset", defaults["market_preset"]))
if market_preset not in MARKET_PRESET_OPTIONS: if market_preset not in MARKET_PRESET_OPTIONS:
market_preset = str(defaults["market_preset"]) market_preset = str(defaults["market_preset"])
compare_symbols = _normalize_symbol_list(raw.get("compare_symbols", defaults["compare_symbols"]), limit=12)
replay_bars = _clamp_int(raw.get("replay_bars"), fallback=int(defaults["replay_bars"]), minimum=20, maximum=1000)
backtest_min_hold_bars = _clamp_int(
raw.get("backtest_min_hold_bars"),
fallback=int(defaults["backtest_min_hold_bars"]),
minimum=1,
maximum=20,
)
backtest_max_hold_bars = _clamp_int(
raw.get("backtest_max_hold_bars"),
fallback=int(defaults["backtest_max_hold_bars"]),
minimum=1,
maximum=40,
)
backtest_max_hold_bars = max(backtest_max_hold_bars, backtest_min_hold_bars)
return { return {
"symbol": symbol, "symbol": symbol,
@ -159,4 +211,42 @@ def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]:
), ),
"watchlist": watchlist, "watchlist": watchlist,
"market_preset": market_preset, "market_preset": market_preset,
"compare_symbols": compare_symbols,
"enable_compare_symbols": _to_bool(
raw.get("enable_compare_symbols"), fallback=bool(defaults["enable_compare_symbols"])
),
"enable_multi_tf_confirmation": _to_bool(
raw.get("enable_multi_tf_confirmation"),
fallback=bool(defaults["enable_multi_tf_confirmation"]),
),
"advanced_auto_run": _to_bool(raw.get("advanced_auto_run"), fallback=bool(defaults["advanced_auto_run"])),
"enable_regime_filter": _to_bool(
raw.get("enable_regime_filter"), fallback=bool(defaults["enable_regime_filter"])
),
"replay_mode_enabled": _to_bool(
raw.get("replay_mode_enabled"), fallback=bool(defaults["replay_mode_enabled"])
),
"replay_bars": replay_bars,
"enable_alert_rules": _to_bool(raw.get("enable_alert_rules"), fallback=bool(defaults["enable_alert_rules"])),
"alert_on_bull": _to_bool(raw.get("alert_on_bull"), fallback=bool(defaults["alert_on_bull"])),
"alert_on_bear": _to_bool(raw.get("alert_on_bear"), fallback=bool(defaults["alert_on_bear"])),
"alert_webhook_url": str(raw.get("alert_webhook_url", defaults["alert_webhook_url"])).strip(),
"backtest_slippage_bps": round(
_clamp_float(raw.get("backtest_slippage_bps"), fallback=0.0, minimum=0.0, maximum=100.0),
1,
),
"backtest_fee_bps": round(
_clamp_float(raw.get("backtest_fee_bps"), fallback=0.0, minimum=0.0, maximum=100.0),
1,
),
"backtest_stop_loss_pct": round(
_clamp_float(raw.get("backtest_stop_loss_pct"), fallback=0.0, minimum=0.0, maximum=25.0),
2,
),
"backtest_take_profit_pct": round(
_clamp_float(raw.get("backtest_take_profit_pct"), fallback=0.0, minimum=0.0, maximum=25.0),
2,
),
"backtest_min_hold_bars": backtest_min_hold_bars,
"backtest_max_hold_bars": backtest_max_hold_bars,
} }

View File

@ -77,9 +77,9 @@ def render_sidebar(active_profile: str, query_params: Any) -> dict[str, Any]:
"Preset", "Preset",
MARKET_PRESET_OPTIONS, MARKET_PRESET_OPTIONS,
index=preset_index, index=preset_index,
help="Applies default timeframes and behavior tuned for common workflows.", help="Select a template, then click Apply Preset to populate defaults.",
) )
if market_preset != "Custom": if market_preset != "Custom" and st.button("Apply Preset", use_container_width=True):
effective_defaults = apply_market_preset(effective_defaults, market_preset) effective_defaults = apply_market_preset(effective_defaults, market_preset)
st.caption(f"Preset applied: {market_preset}") st.caption(f"Preset applied: {market_preset}")
@ -155,63 +155,158 @@ def render_sidebar(active_profile: str, query_params: Any) -> dict[str, Any]:
) )
drop_live = st.checkbox("Ignore potentially live last bar", value=bool(effective_defaults["drop_live"])) drop_live = st.checkbox("Ignore potentially live last bar", value=bool(effective_defaults["drop_live"]))
st.header("Classification Filters") with st.expander("Classification Filters", expanded=False):
use_body_range = st.checkbox( use_body_range = st.checkbox(
"Use previous body range (ignore wicks)", "Use previous body range (ignore wicks)",
value=bool(effective_defaults["use_body_range"]), value=bool(effective_defaults["use_body_range"]),
) )
volume_filter_enabled = st.checkbox( volume_filter_enabled = st.checkbox(
"Enable volume filter", "Enable volume filter",
value=bool(effective_defaults["volume_filter_enabled"]), value=bool(effective_defaults["volume_filter_enabled"]),
) )
volume_sma_window = st.slider("Volume SMA window", 2, 100, int(effective_defaults["volume_sma_window"])) volume_sma_window = st.slider("Volume SMA window", 2, 100, int(effective_defaults["volume_sma_window"]))
volume_multiplier = st.slider( volume_multiplier = st.slider(
"Min volume / SMA multiplier", "Min volume / SMA multiplier",
0.1, 0.1,
3.0, 3.0,
float(effective_defaults["volume_multiplier"]), float(effective_defaults["volume_multiplier"]),
0.1, 0.1,
) )
gray_fake = st.checkbox("Gray out fake bars", value=bool(effective_defaults["gray_fake"])) gray_fake = st.checkbox("Gray out fake bars", value=bool(effective_defaults["gray_fake"]))
hide_market_closed_gaps = st.checkbox( hide_market_closed_gaps = st.checkbox(
"Hide market-closed gaps (stocks)", "Hide market-closed gaps (stocks)",
value=bool(effective_defaults["hide_market_closed_gaps"]), value=bool(effective_defaults["hide_market_closed_gaps"]),
) )
st.header("Training & Guidance") with st.expander("Training & Guidance", expanded=False):
show_live_guide = st.checkbox( show_live_guide = st.checkbox(
"Show live decision guide", "Show live decision guide",
value=bool(effective_defaults["show_live_guide"]), value=bool(effective_defaults["show_live_guide"]),
help="Shows a plain-English interpretation of current trend state and confirmation status.", help="Shows a plain-English interpretation of current trend state and confirmation status.",
) )
show_past_behavior = st.checkbox( show_past_behavior = st.checkbox(
"Show past behavior examples", "Show past behavior examples",
value=bool(effective_defaults["show_past_behavior"]), value=bool(effective_defaults["show_past_behavior"]),
help="Displays historical example trades based on trend confirmation and reversal signals.", help="Displays historical example trades based on trend confirmation and reversal signals.",
) )
show_trade_markers = st.checkbox( show_trade_markers = st.checkbox(
"Overlay example entries/exits on chart", "Overlay example entries/exits on chart",
value=bool(effective_defaults["show_trade_markers"]), value=bool(effective_defaults["show_trade_markers"]),
help="Adds entry/exit markers for the training examples onto the main chart.", help="Adds entry/exit markers for the training examples onto the main chart.",
) )
focus_chart_on_selected_example = st.checkbox( focus_chart_on_selected_example = st.checkbox(
"Focus chart on selected example", "Focus chart on selected example",
value=bool(effective_defaults["focus_chart_on_selected_example"]), value=bool(effective_defaults["focus_chart_on_selected_example"]),
help="When a training row is selected, zooms chart around that trade window.", help="When a training row is selected, zooms chart around that trade window.",
) )
max_training_examples = st.slider( max_training_examples = st.slider(
"Max training examples", "Max training examples",
5, 5,
100, 100,
int(effective_defaults["max_training_examples"]), int(effective_defaults["max_training_examples"]),
5, 5,
) )
replay_mode_enabled = st.checkbox(
"Replay mode (hide future bars)",
value=bool(effective_defaults["replay_mode_enabled"]),
)
replay_bars = st.slider(
"Replay bars shown",
20,
1000,
int(effective_defaults["replay_bars"]),
5,
)
st.header("Monitoring") with st.expander("Monitoring", expanded=False):
enable_auto_refresh = st.checkbox("Auto-refresh", value=bool(effective_defaults["enable_auto_refresh"])) enable_auto_refresh = st.checkbox("Auto-refresh", value=bool(effective_defaults["enable_auto_refresh"]))
refresh_sec = st.slider("Refresh interval (seconds)", 10, 600, int(effective_defaults["refresh_sec"]), 10) refresh_sec = st.slider("Refresh interval (seconds)", 10, 600, int(effective_defaults["refresh_sec"]), 10)
if enable_auto_refresh: if enable_auto_refresh:
st_autorefresh(interval=refresh_sec * 1000, key="data_refresh") st_autorefresh(interval=refresh_sec * 1000, key="data_refresh")
with st.expander("Advanced Signals", expanded=False):
advanced_auto_run = st.checkbox(
"Auto-run advanced panels (slower)",
value=bool(effective_defaults["advanced_auto_run"]),
)
run_advanced_now = st.button("Run advanced panels now", use_container_width=True)
enable_multi_tf_confirmation = st.checkbox(
"Show multi-timeframe confirmation (1h/4h/1d)",
value=bool(effective_defaults["enable_multi_tf_confirmation"]),
)
enable_regime_filter = st.checkbox(
"Regime filter (stand aside in choppy periods)",
value=bool(effective_defaults["enable_regime_filter"]),
)
with st.expander("Compare Symbols", expanded=False):
enable_compare_symbols = st.checkbox(
"Enable compare symbols panel",
value=bool(effective_defaults["enable_compare_symbols"]),
)
compare_symbols_raw = st.text_input(
"Compare symbols (comma separated)",
value=", ".join(effective_defaults["compare_symbols"]),
help="Example: AAPL, MSFT, NVDA",
)
compare_symbols = _parse_watchlist(compare_symbols_raw)[:12]
with st.expander("Alerts", expanded=False):
enable_alert_rules = st.checkbox(
"Enable alert rules",
value=bool(effective_defaults["enable_alert_rules"]),
)
alert_on_bull = st.checkbox("Alert on bullish confirmations", value=bool(effective_defaults["alert_on_bull"]))
alert_on_bear = st.checkbox("Alert on bearish confirmations", value=bool(effective_defaults["alert_on_bear"]))
alert_webhook_url = st.text_input(
"Webhook URL (optional)",
value=str(effective_defaults["alert_webhook_url"]),
help="Use Zapier/Make/Telegram bot webhook for push delivery.",
)
with st.expander("Backtest Controls", expanded=False):
backtest_slippage_bps = st.slider(
"Slippage (bps per side)",
0.0,
100.0,
float(effective_defaults["backtest_slippage_bps"]),
0.5,
)
backtest_fee_bps = st.slider(
"Fee (bps per side)",
0.0,
100.0,
float(effective_defaults["backtest_fee_bps"]),
0.5,
)
backtest_stop_loss_pct = st.slider(
"Stop loss (%)",
0.0,
25.0,
float(effective_defaults["backtest_stop_loss_pct"]),
0.25,
)
backtest_take_profit_pct = st.slider(
"Take profit (%)",
0.0,
25.0,
float(effective_defaults["backtest_take_profit_pct"]),
0.25,
)
backtest_min_hold_bars = st.slider(
"Min hold bars",
1,
20,
int(effective_defaults["backtest_min_hold_bars"]),
1,
)
backtest_max_hold_bars = st.slider(
"Max hold bars",
backtest_min_hold_bars,
40,
int(max(effective_defaults["backtest_max_hold_bars"], backtest_min_hold_bars)),
1,
)
return { return {
"symbol": symbol, "symbol": symbol,
@ -234,4 +329,22 @@ def render_sidebar(active_profile: str, query_params: Any) -> dict[str, Any]:
"refresh_sec": int(refresh_sec), "refresh_sec": int(refresh_sec),
"watchlist": watchlist, "watchlist": watchlist,
"market_preset": market_preset, "market_preset": market_preset,
"compare_symbols": compare_symbols,
"enable_compare_symbols": bool(enable_compare_symbols),
"enable_multi_tf_confirmation": bool(enable_multi_tf_confirmation),
"advanced_auto_run": bool(advanced_auto_run),
"run_advanced_now": bool(run_advanced_now),
"enable_regime_filter": bool(enable_regime_filter),
"replay_mode_enabled": bool(replay_mode_enabled),
"replay_bars": int(replay_bars),
"enable_alert_rules": bool(enable_alert_rules),
"alert_on_bull": bool(alert_on_bull),
"alert_on_bear": bool(alert_on_bear),
"alert_webhook_url": str(alert_webhook_url).strip(),
"backtest_slippage_bps": float(backtest_slippage_bps),
"backtest_fee_bps": float(backtest_fee_bps),
"backtest_stop_loss_pct": float(backtest_stop_loss_pct),
"backtest_take_profit_pct": float(backtest_take_profit_pct),
"backtest_min_hold_bars": int(backtest_min_hold_bars),
"backtest_max_hold_bars": int(backtest_max_hold_bars),
} }