mbrucedogs/maneshtrader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Matt Bruce 2026-02-16 22:01:06 -06:00
parent 1f32527658
commit 1e73d49aa1
20 changed files with 1358 additions and 1027 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
web/src/ONBOARDING.md
View File

@ -71,27 +71,30 @@ Tip: sign and notarize before sharing broadly, so macOS trust prompts are reduce
1. At app start, complete `Profile Login`:
- `Login profile name`: type an existing profile, then click `Login`
- `PIN (if enabled)`: enter PIN for protected profiles
- `Remember me on this browser`: keeps non-PIN profiles signed in across visits
- `Create profile name`: type a new unique profile, then click `Create Profile`
- `Set PIN (optional, 4-6 digits)`: add lightweight profile protection
2. After login, use sidebar `Switch profile` to change users.
3. Note: profile names are case-insensitive for uniqueness (`Matt` and `matt` count as the same profile).
4. Sidebar `Profile` section shows audit stamps for created time, last login, last settings update, and last symbol.
5. Set `Symbol` (examples: `AAPL`, `MSFT`, `BTC-USD`, `ETH-USD`).
6. Set `Timeframe` (start with `1d` to avoid noisy intraday data).
7. Set `Period` (try `6mo` initially).
8. Keep `Ignore potentially live last bar` ON.
9. Keep filters OFF for baseline:
6. Optional: choose `Market Preset` (`Stocks Swing` or `Crypto Intraday`) for one-click defaults.
7. Optional: add symbols to `Watchlist` and use quick-select buttons.
8. Set `Timeframe` (start with `1d` to avoid noisy intraday data).
9. Set `Period` (try `6mo` initially).
10. Keep `Ignore potentially live last bar` ON.
11. Keep filters OFF for baseline:
- `Use previous body range (ignore wicks)` OFF
- `Enable volume filter` OFF
- `Hide market-closed gaps (stocks)` ON
10. Review top metrics:
12. Review top metrics:
- Current Trend
- Real Bullish Bars
- Real Bearish Bars
- Fake Bars
11. Read `Trend Events` for starts and reversals.
12. Use `Live Decision Guide` to translate trend state into a practical bias/action/invalidation workflow.
13. Keep `Show past behavior examples` ON while learning to review historical entry/exit outcomes.
13. Read `Trend Events` for starts and reversals.
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.
## 5) How To Read The Chart
- Candle layer: full price action
@ -236,10 +239,18 @@ streamlit run app.py --server.port 8502
- Share profile-specific links by including `?profile=<your-name>` in the app URL.
- If you were inactive for more than 30 minutes, the app will require login again.
### Watchlist or preset not applying as expected
- Watchlist uses comma-separated symbols and deduplicates automatically.
- Preset `Custom` disables forced defaults; other presets re-apply tuned defaults each run.
### PIN login fails
- Ensure the profile name is correct (name matching is case-insensitive).
- PIN must be exactly what was set during profile creation (4-6 digits).
### Why did auto-login not happen?
- `Remember me` only auto-restores profiles without PIN.
- PIN-protected profiles still require PIN on a fresh/expired session.
### I still see some time gaps
- For stocks, keep `Hide market-closed gaps (stocks)` ON.
- Daily charts remove weekends; intraday removes weekends + closed hours.

13
web/src/PRD.md
View File

@ -17,9 +17,11 @@ Provide an analysis-only charting tool that classifies OHLC bars as real/fake, t
## 3. Inputs and Data Pipeline
1. User configures:
- `symbol`
- optional `watchlist` (per-profile)
- `interval`
- `period`
- `max_bars`
- optional `market_preset`
- filter/toggle settings
2. App fetches OHLCV via Yahoo Finance (`yfinance`).
3. Optional last-bar drop (live-bar guard) for intraday intervals.
@ -46,6 +48,7 @@ Profile behavior:
- App enforces a profile login gate before data/settings UI is shown.
- Login requires typing an existing profile name.
- Login checks PIN when profile has one configured.
- Login includes `Remember me on this browser`.
- Create Profile requires typing a new profile name.
- Create Profile accepts optional 4-6 digit PIN.
- Profile-name uniqueness is case-insensitive (for example `Matt` and `matt` are treated as duplicates).
@ -55,6 +58,14 @@ Profile behavior:
- Save/load are scoped to active profile to avoid cross-user overwrites.
- If profile has no saved settings, defaults are used.
- Session activity is tracked in-memory; after 30 minutes of inactivity the app clears active profile and returns to login screen.
- If `remember=1` is present with a known non-PIN profile, app auto-restores that profile without showing login.
- PIN-protected profiles always require PIN entry after session timeout/reopen.
Watchlist and preset behavior:
- `watchlist` is saved per profile and supports comma/newline input.
- Watchlist supports one-click symbol quick-select buttons.
- `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.
Normalization constraints:
- `symbol`: uppercase, non-empty fallback `AAPL`
@ -69,6 +80,8 @@ Normalization constraints:
- `show_trade_markers`: boolean, fallback `false`
- `focus_chart_on_selected_example`: boolean, fallback `false`
- `max_training_examples`: `[5, 100]`, fallback `20`
- `watchlist`: uppercase de-duplicated symbol list, max 40 items
- `market_preset`: one of `Custom`, `Stocks Swing`, `Crypto Intraday`, `Crypto Swing`, fallback `Custom`
- booleans normalized from common truthy/falsy strings and numbers
## 5. Classification Rules

1028
web/src/app.py
View File

File diff suppressed because it is too large Load Diff

110
web/src/tests/test_app_settings.py
View File

@ -1,23 +1,22 @@
from __future__ import annotations
import json
from pathlib import Path
import app
from web_core.auth import profile_store as store
def test_save_and_load_settings_are_isolated_by_profile(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
app.save_web_settings({"symbol": "AAPL", "period": "1mo"}, profile_id="alice")
app.save_web_settings({"symbol": "MSFT", "period": "3mo"}, profile_id="bob")
store.save_web_settings({"symbol": "AAPL", "period": "1mo"}, profile_id="alice")
store.save_web_settings({"symbol": "MSFT", "period": "3mo"}, profile_id="bob")
alice = app.load_web_settings(profile_id="alice")
bob = app.load_web_settings(profile_id="bob")
unknown = app.load_web_settings(profile_id="carol")
alice = store.load_web_settings(profile_id="alice")
bob = store.load_web_settings(profile_id="bob")
unknown = store.load_web_settings(profile_id="carol")
assert alice["symbol"] == "AAPL"
assert alice["period"] == "1mo"
@ -29,18 +28,18 @@ def test_save_and_load_settings_are_isolated_by_profile(tmp_path, monkeypatch) -
def test_load_settings_migrates_legacy_flat_payload_to_default_profile(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
legacy_payload = {"symbol": "TSLA", "interval": "1h", "max_bars": 9999}
legacy_path.write_text(json.dumps(legacy_payload), encoding="utf-8")
loaded = app.load_web_settings(profile_id="default")
loaded = store.load_web_settings(profile_id="default")
assert loaded["symbol"] == "TSLA"
assert loaded["interval"] == "1h"
assert loaded["max_bars"] == 5000 # normalized max cap
app.save_web_settings({"symbol": "TSLA"}, profile_id="default")
store.save_web_settings({"symbol": "TSLA"}, profile_id="default")
stored = json.loads(settings_path.read_text(encoding="utf-8"))
assert "profiles" in stored
assert "default" in stored["profiles"]
@ -51,68 +50,76 @@ def test_load_settings_migrates_legacy_flat_payload_to_default_profile(tmp_path,
def test_list_web_profiles_includes_saved_profiles(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
app.save_web_settings({"symbol": "AAPL"}, profile_id="zoe")
app.save_web_settings({"symbol": "AAPL"}, profile_id="alice")
store.save_web_settings({"symbol": "AAPL"}, profile_id="zoe")
store.save_web_settings({"symbol": "AAPL"}, profile_id="alice")
profile_ids = app.list_web_profiles()
profile_ids = store.list_web_profiles()
assert "default" in profile_ids
assert "alice" in profile_ids
assert "zoe" in profile_ids
def test_resolve_login_profile_prefers_session_then_query() -> None:
assert app.resolve_login_profile("alice", "bob") == "alice"
assert app.resolve_login_profile("", "bob") == "bob"
assert app.resolve_login_profile(None, None) is None
assert store.resolve_login_profile("alice", "bob") == "alice"
assert store.resolve_login_profile("", "bob") == "bob"
assert store.resolve_login_profile(None, None) is None
def test_profile_exists_normalizes_input() -> None:
profiles = {"default", "alice"}
assert app.profile_exists("alice", profiles) is True
assert app.profile_exists(" ", profiles) is True # normalizes to default
assert app.profile_exists("bob", profiles) is False
assert store.profile_exists("alice", profiles) is True
assert store.profile_exists(" ", profiles) is True # normalizes to default
assert store.profile_exists("bob", profiles) is False
def test_profile_exists_is_case_insensitive() -> None:
profiles = {"default", "Alice"}
assert app.profile_exists("alice", profiles) is True
assert app.find_existing_profile_id("ALICE", profiles) == "Alice"
assert store.profile_exists("alice", profiles) is True
assert store.find_existing_profile_id("ALICE", profiles) == "Alice"
def test_is_profile_session_expired_after_timeout() -> None:
now_epoch = 2000.0
assert app.is_profile_session_expired(1000.0, now_epoch, timeout_sec=900) is True
assert app.is_profile_session_expired(1500.0, now_epoch, timeout_sec=900) is False
assert app.is_profile_session_expired(None, now_epoch, timeout_sec=900) is False
assert store.is_profile_session_expired(1000.0, now_epoch, timeout_sec=900) is True
assert store.is_profile_session_expired(1500.0, now_epoch, timeout_sec=900) is False
assert store.is_profile_session_expired(None, now_epoch, timeout_sec=900) is False
def test_is_truthy_flag_parses_common_values() -> None:
assert store.is_truthy_flag("1") is True
assert store.is_truthy_flag("true") is True
assert store.is_truthy_flag("yes") is True
assert store.is_truthy_flag("0") is False
assert store.is_truthy_flag(None) is False
def test_create_profile_with_pin_requires_pin_for_verification(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
app.create_profile("alice", pin="1234", now_epoch=1700000000)
store.create_profile("alice", pin="1234", now_epoch=1700000000)
assert app.profile_requires_pin("alice") is True
assert app.verify_profile_pin("alice", "1234") is True
assert app.verify_profile_pin("alice", "9999") is False
assert store.profile_requires_pin("alice") is True
assert store.verify_profile_pin("alice", "1234") is True
assert store.verify_profile_pin("alice", "9999") is False
def test_audit_stamps_update_on_save_and_login(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
app.create_profile("alice", now_epoch=1700000000)
app.save_web_settings({"symbol": "MSFT"}, profile_id="alice")
app.mark_profile_login("alice", now_epoch=1700000010)
store.create_profile("alice", now_epoch=1700000000)
store.save_web_settings({"symbol": "MSFT"}, profile_id="alice")
store.mark_profile_login("alice", now_epoch=1700000010)
audit = app.get_profile_audit("alice")
audit = store.get_profile_audit("alice")
assert audit["created_at"] == 1700000000
assert audit["last_login_at"] == 1700000010
assert audit["last_symbol"] == "MSFT"
@ -121,8 +128,8 @@ def test_audit_stamps_update_on_save_and_login(tmp_path, monkeypatch) -> None:
def test_load_settings_migrates_old_profiles_map_structure(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(app, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(app, "LEGACY_SETTINGS_PATH", legacy_path)
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
old_payload = {
"last_profile": "alice",
@ -133,12 +140,23 @@ def test_load_settings_migrates_old_profiles_map_structure(tmp_path, monkeypatch
}
settings_path.write_text(json.dumps(old_payload), encoding="utf-8")
alice_settings = app.load_web_settings("alice")
bob_settings = app.load_web_settings("bob")
alice_settings = store.load_web_settings("alice")
bob_settings = store.load_web_settings("bob")
assert alice_settings["symbol"] == "TSLA"
assert bob_settings["symbol"] == "MSFT"
app.save_web_settings({"symbol": "AAPL"}, profile_id="alice")
store.save_web_settings({"symbol": "AAPL"}, profile_id="alice")
stored = json.loads(settings_path.read_text(encoding="utf-8"))
assert isinstance(stored["profiles"]["alice"], dict)
assert "settings" in stored["profiles"]["alice"]
def test_watchlist_round_trip_keeps_multiple_symbols(tmp_path, monkeypatch) -> None:
settings_path = tmp_path / "settings.json"
legacy_path = tmp_path / "legacy.json"
monkeypatch.setattr(store, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(store, "LEGACY_SETTINGS_PATH", legacy_path)
store.save_web_settings({"symbol": "AMD", "watchlist": ["AMD", "TSLA"]}, profile_id="matt")
loaded = store.load_web_settings(profile_id="matt")
assert loaded["watchlist"] == ["AMD", "TSLA"]

31
web/src/tests/test_settings_schema.py Normal file
View File

@ -0,0 +1,31 @@
from __future__ import annotations
from web_core.settings.settings_schema import normalize_web_settings
def test_normalize_watchlist_from_string_and_deduplicates() -> None:
raw = {"watchlist": "aapl, msft\nAAPL\n btc-usd "}
out = normalize_web_settings(raw)
assert out["watchlist"] == ["AAPL", "MSFT", "BTC-USD"]
def test_normalize_watchlist_from_list_caps_length() -> None:
raw_list = [f"sym{i}" for i in range(60)]
out = normalize_web_settings({"watchlist": raw_list})
assert len(out["watchlist"]) == 40
assert out["watchlist"][0] == "SYM0"
def test_market_preset_defaults_to_custom_on_invalid_value() -> None:
out = normalize_web_settings({"market_preset": "not-real"})
assert out["market_preset"] == "Custom"
def test_normalize_watchlist_splits_legacy_escaped_newline_values() -> None:
out = normalize_web_settings({"watchlist": "AAPL\\nMSFT\\NTSLA"})
assert out["watchlist"] == ["AAPL", "MSFT", "TSLA"]
def test_normalize_watchlist_splits_escaped_newlines_inside_list_items() -> None:
out = normalize_web_settings({"watchlist": ["AMD\\NTSLA"]})
assert out["watchlist"] == ["AMD", "TSLA"]

1
web/src/web_core/auth/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Authentication and profile persistence for web app."""

72
web/src/web_core/auth/profile_auth.py Normal file
View File

@ -0,0 +1,72 @@
from __future__ import annotations
import hashlib
from typing import Any
DEFAULT_PROFILE_ID = "default"
PROFILE_SESSION_TIMEOUT_SEC = 1800
def normalize_profile_id(value: Any) -> str:
profile_id = str(value or "").strip()
return profile_id if profile_id else DEFAULT_PROFILE_ID
def _profile_key(value: Any) -> str:
return normalize_profile_id(value).casefold()
def normalize_pin(value: Any) -> str | None:
pin = str(value or "").strip()
if not pin:
return None
if not pin.isdigit():
return None
if len(pin) < 4 or len(pin) > 6:
return None
return pin
def hash_profile_pin(profile_id: str, pin: str) -> str:
digest_input = f"{_profile_key(profile_id)}:{pin}"
return hashlib.sha256(digest_input.encode("utf-8")).hexdigest()
def first_query_param_value(query_params: Any, key: str) -> str | None:
raw = query_params.get(key)
if raw is None:
return None
if isinstance(raw, list):
return str(raw[0]) if raw else None
return str(raw)
def is_truthy_flag(value: Any) -> bool:
normalized = str(value or "").strip().lower()
return normalized in {"1", "true", "yes", "y", "on"}
def resolve_login_profile(session_profile: Any, query_profile: Any) -> str | None:
if str(session_profile or "").strip():
return normalize_profile_id(session_profile)
if str(query_profile or "").strip():
return normalize_profile_id(query_profile)
return None
def find_existing_profile_id(profile_id: Any, available_profiles: set[str]) -> str | None:
requested_key = _profile_key(profile_id)
for existing in available_profiles:
if _profile_key(existing) == requested_key:
return existing
return None
def profile_exists(profile_id: Any, available_profiles: set[str]) -> bool:
return find_existing_profile_id(profile_id, available_profiles) is not None
def is_profile_session_expired(last_active: Any, now_epoch: float, timeout_sec: int = PROFILE_SESSION_TIMEOUT_SEC) -> bool:
if not isinstance(last_active, (int, float)):
return False
return (now_epoch - float(last_active)) > timeout_sec

226
web/src/web_core/auth/profile_store.py Normal file
View File

@ -0,0 +1,226 @@
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any
import pandas as pd
from web_core.auth.profile_auth import (
DEFAULT_PROFILE_ID,
PROFILE_SESSION_TIMEOUT_SEC,
find_existing_profile_id,
first_query_param_value,
hash_profile_pin,
is_truthy_flag,
is_profile_session_expired,
normalize_pin,
normalize_profile_id,
profile_exists,
resolve_login_profile,
)
from web_core.settings.settings_schema import normalize_web_settings
SETTINGS_PATH = Path.home() / ".web_local_shell" / "settings.json"
LEGACY_SETTINGS_PATH = Path.home() / ".manesh_trader" / "settings.json"
def _normalize_epoch(value: Any) -> int | None:
if not isinstance(value, (int, float)):
return None
parsed = int(value)
return parsed if parsed >= 0 else None
def format_epoch(value: Any) -> str:
parsed = _normalize_epoch(value)
if parsed is None:
return "n/a"
try:
return pd.to_datetime(parsed, unit="s", utc=True).strftime("%Y-%m-%d %H:%M UTC")
except Exception:
return "n/a"
def _load_raw_settings_payload() -> dict[str, Any] | None:
source_path = SETTINGS_PATH if SETTINGS_PATH.exists() else LEGACY_SETTINGS_PATH
if not source_path.exists():
return None
try:
payload = json.loads(source_path.read_text(encoding="utf-8"))
return payload if isinstance(payload, dict) else None
except Exception:
return None
def normalize_profile_record(profile_id: str, raw_profile_data: Any) -> dict[str, Any]:
now_epoch = int(time.time())
fallback_settings = normalize_web_settings(None)
settings: dict[str, Any]
pin_hash = ""
raw_audit: Any = {}
if isinstance(raw_profile_data, dict) and isinstance(raw_profile_data.get("settings"), dict):
settings = normalize_web_settings(raw_profile_data.get("settings"))
pin_hash = str(raw_profile_data.get("pin_hash") or "").strip()
if not pin_hash and isinstance(raw_profile_data.get("auth"), dict):
pin_hash = str(raw_profile_data["auth"].get("pin_hash") or "").strip()
raw_audit = raw_profile_data.get("audit")
elif isinstance(raw_profile_data, dict):
settings = normalize_web_settings(raw_profile_data)
else:
settings = fallback_settings
created_at = now_epoch
updated_at = now_epoch
last_login_at = None
last_symbol = str(settings.get("symbol", fallback_settings["symbol"]))
if isinstance(raw_audit, dict):
created_at = _normalize_epoch(raw_audit.get("created_at")) or now_epoch
updated_at = _normalize_epoch(raw_audit.get("updated_at")) or created_at
last_login_at = _normalize_epoch(raw_audit.get("last_login_at"))
raw_last_symbol = str(raw_audit.get("last_symbol") or "").strip().upper()
if raw_last_symbol:
last_symbol = raw_last_symbol
return {
"settings": settings,
"pin_hash": pin_hash,
"audit": {
"created_at": created_at,
"updated_at": updated_at,
"last_login_at": last_login_at,
"last_symbol": last_symbol,
},
}
def _write_settings_store(profile_records: dict[str, dict[str, Any]], last_profile: str) -> None:
serialized_profiles: dict[str, dict[str, Any]] = {}
for profile_id, record in profile_records.items():
serialized_profiles[profile_id] = normalize_profile_record(profile_id=profile_id, raw_profile_data=record)
payload = {
"last_profile": normalize_profile_id(last_profile),
"profiles": serialized_profiles,
}
SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
SETTINGS_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
def _load_profile_records() -> tuple[dict[str, dict[str, Any]], str]:
payload = _load_raw_settings_payload()
if not payload:
return {}, DEFAULT_PROFILE_ID
raw_profiles = payload.get("profiles")
if isinstance(raw_profiles, dict):
profile_records: dict[str, dict[str, Any]] = {}
for raw_profile_id, raw_profile_data in raw_profiles.items():
if isinstance(raw_profile_data, dict):
profile_id = normalize_profile_id(raw_profile_id)
profile_records[profile_id] = normalize_profile_record(profile_id, raw_profile_data)
last_profile = normalize_profile_id(payload.get("last_profile"))
if profile_records and last_profile not in profile_records:
last_profile = next(iter(profile_records))
return profile_records, last_profile
legacy_record = normalize_profile_record(profile_id=DEFAULT_PROFILE_ID, raw_profile_data=payload)
return {DEFAULT_PROFILE_ID: legacy_record}, DEFAULT_PROFILE_ID
def list_web_profiles() -> list[str]:
profile_records, _ = _load_profile_records()
profile_ids = list(profile_records.keys())
if DEFAULT_PROFILE_ID not in profile_ids:
profile_ids.append(DEFAULT_PROFILE_ID)
return sorted(profile_ids, key=lambda item: (item != DEFAULT_PROFILE_ID, item.lower()))
def load_web_settings(profile_id: str | None = None) -> dict[str, Any]:
profile_records, last_profile = _load_profile_records()
selected_profile = normalize_profile_id(profile_id if profile_id is not None else last_profile)
if selected_profile not in profile_records:
return normalize_web_settings(None)
return normalize_web_settings(profile_records[selected_profile].get("settings"))
def profile_requires_pin(profile_id: str) -> bool:
selected = normalize_profile_id(profile_id)
profile_records, _ = _load_profile_records()
record = profile_records.get(selected)
return isinstance(record, dict) and bool(str(record.get("pin_hash") or "").strip())
def verify_profile_pin(profile_id: str, pin: Any) -> bool:
selected = normalize_profile_id(profile_id)
profile_records, _ = _load_profile_records()
record = profile_records.get(selected)
if not isinstance(record, dict):
return False
stored_hash = str(record.get("pin_hash") or "").strip()
if not stored_hash:
return True
normalized_pin = normalize_pin(pin)
if normalized_pin is None:
return False
return stored_hash == hash_profile_pin(selected, normalized_pin)
def get_profile_audit(profile_id: str) -> dict[str, Any]:
selected = normalize_profile_id(profile_id)
profile_records, _ = _load_profile_records()
record = profile_records.get(selected)
if not isinstance(record, dict):
return {}
audit = record.get("audit")
return audit if isinstance(audit, dict) else {}
def create_profile(profile_id: str, pin: Any = None, now_epoch: int | None = None) -> None:
selected = normalize_profile_id(profile_id)
profile_records, _ = _load_profile_records()
epoch = int(now_epoch if now_epoch is not None else time.time())
normalized_pin = normalize_pin(pin)
profile_records[selected] = {
"settings": normalize_web_settings(None),
"pin_hash": hash_profile_pin(selected, normalized_pin) if normalized_pin else "",
"audit": {
"created_at": epoch,
"updated_at": epoch,
"last_login_at": epoch,
"last_symbol": "AAPL",
},
}
_write_settings_store(profile_records, last_profile=selected)
def mark_profile_login(profile_id: str, now_epoch: int | None = None) -> None:
selected = normalize_profile_id(profile_id)
profile_records, _ = _load_profile_records()
epoch = int(now_epoch if now_epoch is not None else time.time())
existing = profile_records.get(selected, normalize_profile_record(selected, {}))
normalized = normalize_profile_record(selected, existing)
normalized["audit"]["last_login_at"] = epoch
profile_records[selected] = normalized
_write_settings_store(profile_records, last_profile=selected)
def save_web_settings(settings: dict[str, Any], profile_id: str | None = None) -> None:
selected_profile = normalize_profile_id(profile_id)
now_epoch = int(time.time())
profile_records, _ = _load_profile_records()
existing = profile_records.get(selected_profile, normalize_profile_record(selected_profile, {}))
normalized_existing = normalize_profile_record(selected_profile, existing)
normalized_settings = normalize_web_settings(settings)
normalized_existing["settings"] = normalized_settings
normalized_existing["audit"]["updated_at"] = now_epoch
normalized_existing["audit"]["last_symbol"] = str(normalized_settings.get("symbol", "AAPL"))
profile_records[selected_profile] = normalized_existing
_write_settings_store(profile_records, last_profile=selected_profile)

112
web/src/web_core/chart_overlays.py Normal file
View File

@ -0,0 +1,112 @@
from __future__ import annotations
import pandas as pd
import plotly.graph_objects as go
def add_example_trade_markers(fig: go.Figure, example_trades: pd.DataFrame) -> None:
if example_trades.empty:
return
long_entries = example_trades[example_trades["direction"] == "LONG"]
short_entries = example_trades[example_trades["direction"] == "SHORT"]
win_exits = example_trades[example_trades["outcome"] == "Win"]
non_win_exits = example_trades[example_trades["outcome"] != "Win"]
if not long_entries.empty:
fig.add_trace(
go.Scatter(
x=long_entries["entry_timestamp"],
y=long_entries["entry_price"],
mode="markers",
name="Example Entry (Long)",
marker=dict(color="#1565C0", size=9, symbol="circle"),
),
row=1,
col=1,
)
if not short_entries.empty:
fig.add_trace(
go.Scatter(
x=short_entries["entry_timestamp"],
y=short_entries["entry_price"],
mode="markers",
name="Example Entry (Short)",
marker=dict(color="#EF6C00", size=9, symbol="diamond"),
),
row=1,
col=1,
)
if not win_exits.empty:
fig.add_trace(
go.Scatter(
x=win_exits["exit_timestamp"],
y=win_exits["exit_price"],
mode="markers",
name="Example Exit (Win)",
marker=dict(color="#2E7D32", size=10, symbol="x"),
),
row=1,
col=1,
)
if not non_win_exits.empty:
fig.add_trace(
go.Scatter(
x=non_win_exits["exit_timestamp"],
y=non_win_exits["exit_price"],
mode="markers",
name="Example Exit (Loss/Flat)",
marker=dict(color="#C62828", size=10, symbol="x"),
),
row=1,
col=1,
)
def highlight_selected_trade(
fig: go.Figure,
analyzed: pd.DataFrame,
selected_trade: pd.Series | None,
focus_chart_on_selected_example: bool,
) -> None:
if selected_trade is None:
return
selected_entry_ts = pd.Timestamp(selected_trade["entry_timestamp"])
selected_exit_ts = pd.Timestamp(selected_trade["exit_timestamp"])
selected_entry_price = float(selected_trade["entry_price"])
selected_exit_price = float(selected_trade["exit_price"])
selected_direction = str(selected_trade["direction"])
selected_outcome = str(selected_trade["outcome"])
path_color = "#43A047" if selected_outcome == "Win" else ("#EF6C00" if selected_outcome == "Flat" else "#E53935")
window_fill = "#BBDEFB" if selected_direction == "LONG" else "#FFE0B2"
fig.add_vrect(
x0=selected_entry_ts,
x1=selected_exit_ts,
fillcolor=window_fill,
opacity=0.18,
line_width=0,
row=1,
col=1,
)
fig.add_trace(
go.Scatter(
x=[selected_entry_ts, selected_exit_ts],
y=[selected_entry_price, selected_exit_price],
mode="lines+markers",
name="Selected Example Path",
line=dict(color=path_color, width=3, dash="dot"),
marker=dict(color=path_color, size=11, symbol="star"),
),
row=1,
col=1,
)
if focus_chart_on_selected_example:
entry_pos = int(analyzed.index.get_indexer([selected_entry_ts], method="nearest")[0])
exit_pos = int(analyzed.index.get_indexer([selected_exit_ts], method="nearest")[0])
left_pos = max(0, min(entry_pos, exit_pos) - 4)
right_pos = min(len(analyzed) - 1, max(entry_pos, exit_pos) + 4)
fig.update_xaxes(range=[analyzed.index[left_pos], analyzed.index[right_pos]])

45
web/src/web_core/live_guide.py Normal file
View File

@ -0,0 +1,45 @@
from __future__ import annotations
from web_core.constants import TREND_BEAR, TREND_BULL
def build_live_decision_guide(
trend_now: str,
previous_trend: str,
latest_classification: str,
) -> dict[str, str]:
if trend_now == TREND_BULL:
bias = "Long Bias"
action = "Prefer pullback longs and avoid fresh shorts while the bullish trend remains active."
invalidation = "Bullish bias is invalidated only after 2 consecutive real bearish bars confirm a reversal."
elif trend_now == TREND_BEAR:
bias = "Short Bias"
action = "Prefer short setups on pops and avoid fresh longs while the bearish trend remains active."
invalidation = "Bearish bias is invalidated only after 2 consecutive real bullish bars confirm a reversal."
else:
bias = "Stand Aside / Neutral"
action = "Wait for 2 consecutive real bars in one direction before taking directional exposure."
invalidation = "No active trend yet; avoid forcing trades in noisy ranges."
if trend_now in {TREND_BULL, TREND_BEAR} and trend_now != previous_trend:
confirmation = "Fresh Confirmation"
confirmation_detail = "Latest closed bar confirmed a new active trend."
else:
confirmation = "No New Confirmation"
confirmation_detail = "Latest closed bar did not confirm a new trend reversal."
classification_hint = {
"real_bull": "Latest bar closed above the previous range.",
"real_bear": "Latest bar closed below the previous range.",
"fake": "Latest bar stayed inside the previous range (noise).",
"unclassified": "Latest bar is unclassified.",
}.get(latest_classification, "Latest bar classification unavailable.")
return {
"bias": bias,
"action": action,
"invalidation": invalidation,
"confirmation": confirmation,
"confirmation_detail": confirmation_detail,
"classification_hint": classification_hint,
}

1
web/src/web_core/market/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Market-facing helpers (symbols, presets)."""

49
web/src/web_core/market/presets.py Normal file
View File

@ -0,0 +1,49 @@
from __future__ import annotations
from typing import Any
MARKET_PRESET_OPTIONS = [
"Custom",
"Stocks Swing",
"Crypto Intraday",
"Crypto Swing",
]
def apply_market_preset(base: dict[str, Any], preset: str) -> dict[str, Any]:
updated = dict(base)
if preset == "Stocks Swing":
updated.update(
{
"interval": "1d",
"period": "6mo",
"max_bars": 500,
"drop_live": True,
"hide_market_closed_gaps": True,
"enable_auto_refresh": False,
}
)
elif preset == "Crypto Intraday":
updated.update(
{
"interval": "1h",
"period": "3mo",
"max_bars": 1000,
"drop_live": True,
"hide_market_closed_gaps": False,
"enable_auto_refresh": True,
"refresh_sec": 60,
}
)
elif preset == "Crypto Swing":
updated.update(
{
"interval": "4h",
"period": "6mo",
"max_bars": 800,
"drop_live": True,
"hide_market_closed_gaps": False,
"enable_auto_refresh": False,
}
)
return updated

73
web/src/web_core/market/symbols.py Normal file
View File

@ -0,0 +1,73 @@
from __future__ import annotations
from typing import Any
import streamlit as st
import yfinance as yf
@st.cache_data(show_spinner=False, ttl=3600)
def lookup_symbol_candidates(query: str, max_results: int = 10) -> list[dict[str, str]]:
cleaned = query.strip()
if len(cleaned) < 2:
return []
try:
search = yf.Search(cleaned, max_results=max_results)
quotes = getattr(search, "quotes", []) or []
except Exception:
return []
seen_symbols: set[str] = set()
candidates: list[dict[str, str]] = []
for quote in quotes:
symbol = str(quote.get("symbol", "")).strip().upper()
if not symbol or symbol in seen_symbols:
continue
seen_symbols.add(symbol)
candidates.append(
{
"symbol": symbol,
"name": str(quote.get("shortname") or quote.get("longname") or "").strip(),
"exchange": str(quote.get("exchDisp") or quote.get("exchange") or "").strip(),
"type": str(quote.get("typeDisp") or quote.get("quoteType") or "").strip(),
}
)
return candidates
@st.cache_data(show_spinner=False, ttl=3600)
def resolve_symbol_identity(symbol: str) -> dict[str, str]:
normalized_symbol = symbol.strip().upper()
if not normalized_symbol:
return {"symbol": "", "name": "", "exchange": ""}
def _from_quote(quote: dict[str, Any]) -> dict[str, str]:
return {
"symbol": normalized_symbol,
"name": str(quote.get("shortname") or quote.get("longname") or "").strip(),
"exchange": str(quote.get("exchDisp") or quote.get("exchange") or "").strip(),
}
try:
search = yf.Search(normalized_symbol, max_results=8)
quotes = getattr(search, "quotes", []) or []
for quote in quotes:
candidate_symbol = str(quote.get("symbol", "")).strip().upper()
if candidate_symbol == normalized_symbol:
return _from_quote(quote)
if quotes:
return _from_quote(quotes[0])
except Exception:
pass
try:
info = yf.Ticker(normalized_symbol).info
return {
"symbol": normalized_symbol,
"name": str(info.get("shortName") or info.get("longName") or "").strip(),
"exchange": str(info.get("exchange") or "").strip(),
}
except Exception:
return {"symbol": normalized_symbol, "name": "", "exchange": ""}

1
web/src/web_core/settings/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Settings schema and normalization utilities."""

162
web/src/web_core/settings/settings_schema.py Normal file
View File

@ -0,0 +1,162 @@
from __future__ import annotations
from typing import Any
from web_core.constants import INTERVAL_OPTIONS, PERIOD_OPTIONS
from web_core.market.presets import MARKET_PRESET_OPTIONS
def _clamp_int(value: Any, fallback: int, minimum: int, maximum: int) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return fallback
return min(maximum, max(minimum, parsed))
def _clamp_float(value: Any, fallback: float, minimum: float, maximum: float) -> float:
try:
parsed = float(value)
except (TypeError, ValueError):
return fallback
return min(maximum, max(minimum, parsed))
def _to_bool(value: Any, fallback: bool) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return value != 0
if value is None:
return fallback
normalized = str(value).strip().lower()
if normalized in {"1", "true", "yes", "y", "on"}:
return True
if normalized in {"0", "false", "no", "n", "off"}:
return False
return fallback
def _clamp_max_bars(value: Any, fallback: int = 500) -> int:
return _clamp_int(value=value, fallback=fallback, minimum=20, maximum=5000)
def _normalize_watchlist(raw: Any) -> list[str]:
tokens: list[str] = []
if isinstance(raw, list):
tokens = [str(item) for item in raw]
elif isinstance(raw, str):
tokens = [raw]
else:
tokens = [str(raw or "")]
seen: set[str] = set()
symbols: 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)
symbols.append(symbol)
return symbols[:40]
def normalize_web_settings(raw: dict[str, Any] | None) -> dict[str, Any]:
raw = raw or {}
defaults: dict[str, Any] = {
"symbol": "AAPL",
"interval": "1d",
"period": "6mo",
"max_bars": 500,
"drop_live": True,
"use_body_range": False,
"volume_filter_enabled": False,
"volume_sma_window": 20,
"volume_multiplier": 1.0,
"gray_fake": True,
"hide_market_closed_gaps": True,
"show_live_guide": False,
"show_past_behavior": False,
"show_trade_markers": False,
"focus_chart_on_selected_example": False,
"max_training_examples": 20,
"enable_auto_refresh": False,
"refresh_sec": 60,
"watchlist": [],
"market_preset": "Custom",
}
symbol = str(raw.get("symbol", defaults["symbol"])).strip().upper()
if not symbol:
symbol = str(defaults["symbol"])
interval = str(raw.get("interval", defaults["interval"]))
if interval not in INTERVAL_OPTIONS:
interval = str(defaults["interval"])
period = str(raw.get("period", defaults["period"]))
if period not in PERIOD_OPTIONS:
period = str(defaults["period"])
watchlist = _normalize_watchlist(raw.get("watchlist", defaults["watchlist"]))
market_preset = str(raw.get("market_preset", defaults["market_preset"]))
if market_preset not in MARKET_PRESET_OPTIONS:
market_preset = str(defaults["market_preset"])
return {
"symbol": symbol,
"interval": interval,
"period": period,
"max_bars": _clamp_max_bars(raw.get("max_bars"), fallback=int(defaults["max_bars"])),
"drop_live": _to_bool(raw.get("drop_live"), fallback=bool(defaults["drop_live"])),
"use_body_range": _to_bool(raw.get("use_body_range"), fallback=bool(defaults["use_body_range"])),
"volume_filter_enabled": _to_bool(
raw.get("volume_filter_enabled"), fallback=bool(defaults["volume_filter_enabled"])
),
"volume_sma_window": _clamp_int(
raw.get("volume_sma_window"),
fallback=int(defaults["volume_sma_window"]),
minimum=2,
maximum=100,
),
"volume_multiplier": round(
_clamp_float(
raw.get("volume_multiplier"),
fallback=float(defaults["volume_multiplier"]),
minimum=0.1,
maximum=3.0,
),
1,
),
"gray_fake": _to_bool(raw.get("gray_fake"), fallback=bool(defaults["gray_fake"])),
"hide_market_closed_gaps": _to_bool(
raw.get("hide_market_closed_gaps"),
fallback=bool(defaults["hide_market_closed_gaps"]),
),
"show_live_guide": _to_bool(raw.get("show_live_guide"), fallback=bool(defaults["show_live_guide"])),
"show_past_behavior": _to_bool(raw.get("show_past_behavior"), fallback=bool(defaults["show_past_behavior"])),
"show_trade_markers": _to_bool(raw.get("show_trade_markers"), fallback=bool(defaults["show_trade_markers"])),
"focus_chart_on_selected_example": _to_bool(
raw.get("focus_chart_on_selected_example"),
fallback=bool(defaults["focus_chart_on_selected_example"]),
),
"max_training_examples": _clamp_int(
raw.get("max_training_examples"),
fallback=int(defaults["max_training_examples"]),
minimum=5,
maximum=100,
),
"enable_auto_refresh": _to_bool(
raw.get("enable_auto_refresh"), fallback=bool(defaults["enable_auto_refresh"])
),
"refresh_sec": _clamp_int(
raw.get("refresh_sec"),
fallback=int(defaults["refresh_sec"]),
minimum=10,
maximum=600,
),
"watchlist": watchlist,
"market_preset": market_preset,
}

1
web/src/web_core/ui/__init__.py Normal file
View File

@ -0,0 +1 @@
"""UI composition helpers for the Streamlit app."""

28
web/src/web_core/ui/help_content.py Normal file
View File

@ -0,0 +1,28 @@
from __future__ import annotations
from pathlib import Path
import streamlit as st
@st.cache_data(show_spinner=False)
def load_help_content() -> tuple[str, bool]:
app_path = Path(__file__).resolve().parents[1] / "app.py"
help_html_paths = [
app_path.with_name("web_core").joinpath("help.html"),
app_path.with_name("help.html"),
]
for help_html_path in help_html_paths:
if help_html_path.exists():
return help_html_path.read_text(encoding="utf-8"), True
onboarding_path = app_path.with_name("ONBOARDING.md")
if onboarding_path.exists():
return onboarding_path.read_text(encoding="utf-8"), False
return "Help content not found.", False
@st.dialog("Help & Quick Start", width="large")
def help_dialog() -> None:
content, is_html = load_help_content()
st.markdown(content, unsafe_allow_html=is_html)

69
web/src/web_core/ui/login_ui.py Normal file
View File

@ -0,0 +1,69 @@
from __future__ import annotations
from typing import Any
import streamlit as st
from web_core.auth.profile_store import (
create_profile,
find_existing_profile_id,
first_query_param_value,
is_truthy_flag,
list_web_profiles,
mark_profile_login,
normalize_pin,
normalize_profile_id,
profile_exists,
profile_requires_pin,
verify_profile_pin,
)
def render_profile_login(now_epoch: float, query_params: Any, session_expired: bool) -> None:
existing_profiles = set(list_web_profiles())
st.subheader("Profile Login")
st.info("Login with an existing profile or create a new one. Settings are isolated per profile.")
if session_expired:
st.warning("Your profile session timed out due to inactivity. Please log in again.")
initial_profile = first_query_param_value(query_params, "profile") or ""
remember_default = is_truthy_flag(first_query_param_value(query_params, "remember"))
login_profile = st.text_input("Login profile name", value=initial_profile, placeholder="e.g. matt")
login_pin = st.text_input("PIN (if enabled)", value="", type="password", max_chars=6)
remember_me = st.checkbox("Remember me on this browser", value=remember_default or True)
if st.button("Login", type="primary", use_container_width=True):
selected_match = find_existing_profile_id(login_profile, existing_profiles)
if selected_match is None:
st.error("Profile not found. Enter the exact profile name or create a new one below.")
elif profile_requires_pin(selected_match) and not verify_profile_pin(selected_match, login_pin):
st.error("Incorrect PIN.")
else:
mark_profile_login(selected_match, now_epoch=int(now_epoch))
st.session_state["active_profile"] = selected_match
st.session_state["profile_last_active_at"] = now_epoch
query_params["profile"] = selected_match
if remember_me:
query_params["remember"] = "1"
elif "remember" in query_params:
del query_params["remember"]
st.rerun()
st.divider()
create_profile_name = st.text_input("Create profile name", value="", placeholder="e.g. sara")
create_pin = st.text_input("Set PIN (optional, 4-6 digits)", value="", type="password", max_chars=6)
if st.button("Create Profile", use_container_width=True):
selected = normalize_profile_id(create_profile_name)
if profile_exists(selected, existing_profiles):
st.error("That profile already exists (including case-insensitive matches). Use Login instead.")
elif create_pin and normalize_pin(create_pin) is None:
st.error("PIN must be 4-6 digits.")
else:
create_profile(selected, pin=create_pin or None, now_epoch=int(now_epoch))
st.session_state["active_profile"] = selected
st.session_state["profile_last_active_at"] = now_epoch
query_params["profile"] = selected
if remember_me:
query_params["remember"] = "1"
elif "remember" in query_params:
del query_params["remember"]
st.rerun()

237
web/src/web_core/ui/sidebar_ui.py Normal file
View File

@ -0,0 +1,237 @@
from __future__ import annotations
from typing import Any
import streamlit as st
from streamlit_autorefresh import st_autorefresh
from web_core.constants import INTERVAL_OPTIONS, PERIOD_OPTIONS
from web_core.ui.help_content import help_dialog
from web_core.market.presets import MARKET_PRESET_OPTIONS, apply_market_preset
from web_core.market.symbols import lookup_symbol_candidates
from web_core.auth.profile_store import (
first_query_param_value,
format_epoch,
get_profile_audit,
list_web_profiles,
load_web_settings,
normalize_web_settings,
)
def _parse_watchlist(raw: str) -> list[str]:
seen: set[str] = set()
symbols: list[str] = []
normalized_raw = str(raw or "").replace("\\n", "\n").replace("\\N", "\n").replace(",", "\n")
for token in normalized_raw.splitlines():
normalized = token.strip().upper()
if not normalized or normalized in seen:
continue
seen.add(normalized)
symbols.append(normalized)
return symbols[:40]
def render_sidebar(active_profile: str, query_params: Any) -> dict[str, Any]:
with st.sidebar:
st.header("Profile")
st.success(f"Logged in as: {active_profile}")
if st.button("Switch profile", use_container_width=True):
st.session_state.pop("active_profile", None)
if "profile" in query_params:
del query_params["profile"]
st.rerun()
st.divider()
available_profiles = list_web_profiles()
profile_audit = get_profile_audit(active_profile)
st.caption(
"Profiles found: "
+ ", ".join(available_profiles + ([active_profile] if active_profile not in available_profiles else []))
)
st.caption(
"Audit: created "
f"{format_epoch(profile_audit.get('created_at'))}, "
f"last login {format_epoch(profile_audit.get('last_login_at'))}, "
f"updated {format_epoch(profile_audit.get('updated_at'))}, "
f"last symbol {str(profile_audit.get('last_symbol') or 'n/a')}"
)
st.divider()
st.header("Data Settings")
if st.button("Help / Quick Start", use_container_width=True):
help_dialog()
st.divider()
persisted_settings = load_web_settings(profile_id=active_profile)
query_overrides: dict[str, Any] = {}
for key in persisted_settings:
candidate = first_query_param_value(query_params, key)
if candidate is not None:
query_overrides[key] = candidate
effective_defaults = normalize_web_settings({**persisted_settings, **query_overrides})
st.subheader("Market Preset")
preset_default = str(effective_defaults["market_preset"])
preset_index = MARKET_PRESET_OPTIONS.index(preset_default) if preset_default in MARKET_PRESET_OPTIONS else 0
market_preset = st.selectbox(
"Preset",
MARKET_PRESET_OPTIONS,
index=preset_index,
help="Applies default timeframes and behavior tuned for common workflows.",
)
if market_preset != "Custom":
effective_defaults = apply_market_preset(effective_defaults, market_preset)
st.caption(f"Preset applied: {market_preset}")
st.subheader("Watchlist")
watchlist_raw = st.text_input(
"Watchlist symbols (comma separated)",
value=", ".join(effective_defaults["watchlist"]),
help="Saved per profile. Example: AAPL, MSFT, NVDA, BTC-USD",
)
watchlist = _parse_watchlist(watchlist_raw)
selected_from_watchlist: str | None = None
if watchlist:
st.caption("Quick select:")
for idx, watch_symbol in enumerate(watchlist[:12]):
if st.button(watch_symbol, use_container_width=True, key=f"watchlist-{watch_symbol}-{idx}"):
selected_from_watchlist = watch_symbol
st.session_state["watchlist_symbol_choice"] = watch_symbol
selected_from_watchlist = st.session_state.get("watchlist_symbol_choice", selected_from_watchlist)
st.subheader("Find Symbol")
symbol_search_query = st.text_input(
"Search by company or ticker",
value="",
placeholder="Apple, Tesla, Bitcoin...",
help="Type a name (e.g. Apple) and select a result to prefill Symbol.",
)
symbol_from_search: str | None = None
if symbol_search_query.strip():
candidates = lookup_symbol_candidates(symbol_search_query)
if candidates:
result_placeholder = "Select a result..."
labels = [result_placeholder] + [
" | ".join(
[
candidate["symbol"],
candidate["name"] or "No name",
candidate["exchange"] or "Unknown exchange",
]
)
for candidate in candidates
]
selected_label = st.selectbox("Matches", labels, index=0)
if selected_label != result_placeholder:
selected_index = labels.index(selected_label) - 1
symbol_from_search = candidates[selected_index]["symbol"]
else:
st.caption("No matches found. Try another company name.")
symbol = st.text_input(
"Symbol",
value=symbol_from_search or selected_from_watchlist or str(effective_defaults["symbol"]),
help="Ticker or pair to analyze, e.g. AAPL, MSFT, BTC-USD.",
).strip().upper()
interval = st.selectbox(
"Timeframe",
INTERVAL_OPTIONS,
index=INTERVAL_OPTIONS.index(str(effective_defaults["interval"])),
help="Bar size for each candle. Shorter intervals are noisier; 1d is a good default.",
)
period = st.selectbox(
"Period",
PERIOD_OPTIONS,
index=PERIOD_OPTIONS.index(str(effective_defaults["period"])),
help="How much history to load for trend analysis.",
)
max_bars = st.number_input(
"Max bars",
min_value=20,
max_value=5000,
value=int(effective_defaults["max_bars"]),
step=10,
help="Limits loaded candles to keep charting responsive. 500 is a solid starting point.",
)
drop_live = st.checkbox("Ignore potentially live last bar", value=bool(effective_defaults["drop_live"]))
st.header("Classification Filters")
use_body_range = st.checkbox(
"Use previous body range (ignore wicks)",
value=bool(effective_defaults["use_body_range"]),
)
volume_filter_enabled = st.checkbox(
"Enable volume filter",
value=bool(effective_defaults["volume_filter_enabled"]),
)
volume_sma_window = st.slider("Volume SMA window", 2, 100, int(effective_defaults["volume_sma_window"]))
volume_multiplier = st.slider(
"Min volume / SMA multiplier",
0.1,
3.0,
float(effective_defaults["volume_multiplier"]),
0.1,
)
gray_fake = st.checkbox("Gray out fake bars", value=bool(effective_defaults["gray_fake"]))
hide_market_closed_gaps = st.checkbox(
"Hide market-closed gaps (stocks)",
value=bool(effective_defaults["hide_market_closed_gaps"]),
)
st.header("Training & Guidance")
show_live_guide = st.checkbox(
"Show live decision guide",
value=bool(effective_defaults["show_live_guide"]),
help="Shows a plain-English interpretation of current trend state and confirmation status.",
)
show_past_behavior = st.checkbox(
"Show past behavior examples",
value=bool(effective_defaults["show_past_behavior"]),
help="Displays historical example trades based on trend confirmation and reversal signals.",
)
show_trade_markers = st.checkbox(
"Overlay example entries/exits on chart",
value=bool(effective_defaults["show_trade_markers"]),
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",
value=bool(effective_defaults["focus_chart_on_selected_example"]),
help="When a training row is selected, zooms chart around that trade window.",
)
max_training_examples = st.slider(
"Max training examples",
5,
100,
int(effective_defaults["max_training_examples"]),
5,
)
st.header("Monitoring")
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)
if enable_auto_refresh:
st_autorefresh(interval=refresh_sec * 1000, key="data_refresh")
return {
"symbol": symbol,
"interval": interval,
"period": period,
"max_bars": int(max_bars),
"drop_live": bool(drop_live),
"use_body_range": bool(use_body_range),
"volume_filter_enabled": bool(volume_filter_enabled),
"volume_sma_window": int(volume_sma_window),
"volume_multiplier": float(volume_multiplier),
"gray_fake": bool(gray_fake),
"hide_market_closed_gaps": bool(hide_market_closed_gaps),
"show_live_guide": bool(show_live_guide),
"show_past_behavior": bool(show_past_behavior),
"show_trade_markers": bool(show_trade_markers),
"focus_chart_on_selected_example": bool(focus_chart_on_selected_example),
"max_training_examples": int(max_training_examples),
"enable_auto_refresh": bool(enable_auto_refresh),
"refresh_sec": int(refresh_sec),
"watchlist": watchlist,
"market_preset": market_preset,
}

83
web/src/web_core/ui/training_ui.py Normal file
View File

@ -0,0 +1,83 @@
from __future__ import annotations
import pandas as pd
import streamlit as st
def render_training_panel(show_past_behavior: bool, example_trades: pd.DataFrame, alert_key: str) -> pd.Series | None:
selected_trade: pd.Series | None = None
if not show_past_behavior:
return selected_trade
st.subheader("Past Behavior Examples (Training)")
if example_trades.empty:
st.info("No closed example trades yet. Expand the period/timeframe to include more trend reversals.")
return selected_trade
wins = int((example_trades["outcome"] == "Win").sum())
losses = int((example_trades["outcome"] == "Loss").sum())
total_examples = int(len(example_trades))
win_rate = round((wins / total_examples) * 100.0, 2) if total_examples else 0.0
avg_pnl = round(float(example_trades["pnl_pct"].mean()), 2)
t1, t2, t3, t4 = st.columns(4)
t1.metric("Closed Examples", total_examples)
t2.metric("Wins / Losses", f"{wins} / {losses}")
t3.metric("Example Win Rate", f"{win_rate}%")
t4.metric("Avg P/L per Example", f"{avg_pnl}%")
latest_example = example_trades.iloc[-1]
st.caption(
"Latest closed example: "
f"{latest_example['direction']} from {latest_example['entry_timestamp']} "
f"to {latest_example['exit_timestamp']} "
f"({latest_example['bars_held']} bars, {latest_example['pnl_pct']}%)."
)
st.caption("Click a row below to highlight that specific example on the chart.")
st.caption("Example method: enter on trend confirmation bar close and exit on opposite trend confirmation.")
display_examples_raw = example_trades.iloc[::-1].reset_index(drop=True)
display_examples = display_examples_raw.copy()
display_examples["entry_timestamp"] = display_examples["entry_timestamp"].astype(str)
display_examples["exit_timestamp"] = display_examples["exit_timestamp"].astype(str)
table_event = st.dataframe(
display_examples,
use_container_width=True,
on_select="rerun",
selection_mode="single-row",
key=f"training-examples-{alert_key}",
)
selected_rows: list[int] = []
try:
selected_rows = list(table_event.selection.rows)
except Exception:
selected_rows = []
selected_row_state_key = f"selected-training-row-{alert_key}"
if selected_rows:
selected_row_idx = int(selected_rows[0])
st.session_state[selected_row_state_key] = selected_row_idx
else:
selected_row_idx = int(st.session_state.get(selected_row_state_key, 0))
if selected_row_idx < 0 or selected_row_idx >= len(display_examples_raw):
selected_row_idx = 0
selected_trade = display_examples_raw.iloc[selected_row_idx]
direction = str(selected_trade["direction"])
entry_ts = str(selected_trade["entry_timestamp"])
exit_ts = str(selected_trade["exit_timestamp"])
entry_price = float(selected_trade["entry_price"])
exit_price = float(selected_trade["exit_price"])
bars_held = int(selected_trade["bars_held"])
pnl_pct = float(selected_trade["pnl_pct"])
outcome = str(selected_trade["outcome"])
pnl_text = f"+{pnl_pct}%" if pnl_pct > 0 else f"{pnl_pct}%"
result_text = "profit" if pnl_pct > 0 else ("loss" if pnl_pct < 0 else "flat result")
st.info(
f"Selected example explained: {direction} entry at {entry_ts} ({entry_price}), "
f"exit at {exit_ts} ({exit_price}), held {bars_held} bars, {result_text} {pnl_text}, outcome: {outcome}."
)
return selected_trade