"""Typer CLI — one of two entry points into the Core Service Layer.
The CLI only orchestrates: it resolves config, builds adapters, calls Core
functions, and formats output. No trading logic lives here, so the Slice 2
daemon and a future web dashboard can call the same Core code.
"""
import functools
import json
import logging
import sys
import time
from collections.abc import Callable, Collection, Sequence
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from enum import StrEnum
from importlib.metadata import version as package_version
from pathlib import Path
import typer
import uvicorn
from prettytable import PrettyTable
from sqlalchemy.engine import make_url
from trader.backtest.engine import BacktestConfig, run_backtest
from trader.backtest.metrics import max_drawdown_pct, total_return_pct, win_loss
from trader.brokers.alpaca import AlpacaBroker
from trader.brokers.base import BrokerAdapter
from trader.chat.service import ChatService
from trader.cli.tables import MISSING, build_table
from trader.concurrency import InstanceAlreadyRunningError, single_instance_lock
from trader.config.loader import (
EffectiveTradingConfig,
assemble_database_url,
load_trading_config,
resolve_trading_config,
)
from trader.config.schema import (
DaemonConfig,
DiscoverySettings,
IdeasScraperConfig,
PipelineConfig,
ReportingConfig,
SimulationSettings,
StrategyConfig,
StrategyMode,
load_daemon_config,
load_ideas_scraper_config,
load_pipeline_config,
load_reporting_config,
load_simulation_config,
load_strategies_config,
load_watchlist_config,
)
from trader.config.settings import Settings
from trader.daemon.backoff import Backoff
from trader.daemon.loop import run_forever
from trader.daemon.preflight import confirm_live_trading, preflight_failures
from trader.daemon.schedule import MarketSchedule
from trader.daemon.signals import ShutdownFlag, install_handlers
from trader.daemon.sleeper import RealSleeper
from trader.decisions import DecisionMemory
from trader.discovery.filters import FilterOutcome
from trader.discovery.health_score import calculate_health_score
from trader.discovery.rank_persistence import (
METRIC_RECOMMENDATION_MEAN,
METRIC_STRONG_BUY,
ConsensusHistoryPoint,
compute_rank_stability,
)
from trader.discovery.scan import ANALYST_SCAN_INDEX, discover_symbols, screen_for_entry
from trader.domain import Bar, MarketClock, Position
from trader.errors import TraderError
from trader.evaluation.backtest_evaluator import BacktestEvaluator
from trader.execution.executor import OrderExecutor
from trader.ideas.scraper import scrape_ideas
from trader.llm.base import LlmProvider
from trader.llm.ollama import DEFAULT_NUM_CTX, OllamaProvider
from trader.logging_setup import configure_logging, safe_config_summary
from trader.marketdata.analysts import AnalystProvider, YFinanceAnalystProvider
from trader.marketdata.base import MarketDataProvider
from trader.marketdata.cache import BarCache
from trader.marketdata.earnings import EarningsProvider, YFinanceEarningsProvider
from trader.marketdata.index_provider import WikipediaIndexProvider
from trader.marketdata.sector import SectorProvider, YFinanceSectorProvider
from trader.marketdata.seed_import import (
SOURCE_IBKR_SEED,
SOURCE_YFINANCE_SEED,
load_feather_bars,
load_ibkr_csv_bars,
)
from trader.marketdata.yfinance_provider import YFinanceProvider
from trader.news.aliases import CompanyAliasResolver
from trader.news.alpaca_news import AlpacaNewsProvider
from trader.news.archive import NewsArchive
from trader.news.base import DEFAULT_NEWS_WINDOW_HOURS, NewsProvider
from trader.news.merged_news import MergedNewsProvider
from trader.news.search import NewsSearchProvider, YFinanceSearchProvider
from trader.news.yfinance_news import YFinanceNewsProvider
from trader.performance.benchmark import (
account_return,
benchmark_return,
excess_return_pct,
)
from trader.persistence.analyst_consensus import AnalystConsensusRepository
from trader.persistence.backtests import BacktestRepository
from trader.persistence.bars import BarRepository
from trader.persistence.db import (
DatabaseError,
create_db_engine,
create_session_factory,
init_db,
)
from trader.persistence.decisions import DecisionRepository
from trader.persistence.deployed_versions import DeployedVersionRepository
from trader.persistence.focus_list import FocusListRepository
from trader.persistence.index_membership import IndexMembershipRepository
from trader.persistence.migrations import current_revision, upgrade_to_head
from trader.persistence.models import RoundTrip, SimPortfolio, SimRoundTrip
from trader.persistence.news_archive import NewsArchiveRepository
from trader.persistence.outcomes import OutcomeRepository, OutcomeSummary
from trader.persistence.simulation import (
ABB_PORTFOLIO_SUFFIX,
EXIT_TRAILING_STOP,
SimulationRepository,
)
from trader.persistence.snapshots import SnapshotRepository
from trader.persistence.trades import TradeRepository
from trader.persistence.watchlist import WatchlistEventRepository
from trader.pipeline.run_once import CycleReport, StrategyRun, run_once
# From the submodules, never from `trader.ranking` itself: that package is
# deliberately a docstring with no re-exports, because a re-export there
# re-enters the package while it is still initializing and is a hard
# `ImportError`. Measured 2026-08-05; see its docstring.
from trader.ranking.base import Ranker
from trader.ranking.llm_ranker import LlmRanker
from trader.replay.run import ReplayRunner, rescore_from_csv
from trader.replay.sample import SampleShortfall, build_sample
from trader.replay.sessions import session_grid
from trader.reporting.concentration import (
DEFAULT_CORRELATION_LOOKBACK_DAYS,
PortfolioCorrelation,
SectorCount,
SectorExposure,
compute_correlations,
sector_distribution_by_count,
sector_exposure,
)
from trader.reporting.equity_curve import equity_curve_points, format_report
from trader.reporting.returns import (
PERCENT_QUANTUM,
avg_daily_return_pct,
cost_basis,
days_held,
position_return_pct,
return_pct,
)
from trader.reporting.web.wiring import build_app
from trader.reporting.window import resolve_report_start
from trader.resilience.retrying import RetryingBroker, RetryingMarketData
from trader.risk.guardrails import Guardrails
from trader.simulation import SimulationRunner
from trader.strategies.registry import build_strategy
from trader.versioning import detect_git_version
app = typer.Typer(help="AI-assisted Alpaca trading app.", no_args_is_help=True)
# Administrative/setup commands live in their own subgroup, separate from the
# read-only and trading commands on `app` directly (issue #90). `init-db` was
# previously a bare top-level command, indistinguishable in `--help` output
# from `positions`/`account`/`quote` — routine, read-mostly commands — even
# though it is schema-administration, not something to run casually. The
# `create_all` call it makes is additive and idempotent (it never drops or
# truncates), so this is a *visibility* fix, not a safety gate in the
# `assert_live_orders_allowed()` sense: nothing here blocks the command from
# running, it just stops appearing next to `account`/`positions` in
# `trader --help`. A config-file opt-in (the operator's other suggested shape)
# would add friction without addressing that — the actual complaint in issue
# #90 is discoverability, not "this command is too easy to run by accident".
setup_app = typer.Typer(
help="Administrative setup commands (schema creation).", no_args_is_help=True
)
app.add_typer(setup_app, name="setup")
_logger = logging.getLogger("trader.cli")
# There is no trend filter (removed in Task 5 of Slice 3d — see the design
# doc's amended item 7), so discovery's bar window only needs to cover what
# the price and volume filters actually read: the last close, and a short
# recent average. 30 calendar days is comfortably enough daily bars for both
# and is deliberately NOT sized for a downtrend lookback that no longer
# exists.
_DISCOVERY_BAR_WINDOW_DAYS = 30
# How far back `--risk` looks in `watchlist_events` for "recently discovered"
# candidates (issue #65's sector-distribution-by-count half). A week is long
# enough to span a quiet news cycle without pulling in a candidate discovery
# considered and dropped a month ago — the audit trail keeps everything, but
# this report describes the *current* candidate pool, not its whole history.
_RISK_DISCOVERED_LOOKBACK_DAYS = 7
@app.callback()
def _configure(ctx: typer.Context) -> None:
"""Runs before every command.
Logging is set up here rather than in each command because it was in three
of fifteen, so `LOG_LEVEL` was a lie for the other twelve. One call site
also means one place to change it.
Unconditional — no `ctx.resilient_parsing` check. An earlier version
special-cased it, believing that guard was what kept `--help` from
creating `logs/`. It was dead code: `ctx.resilient_parsing` is never True
on any path reachable from this app (top-level `--help` short-circuits in
Click before this callback runs at all; `trader run --help` reaches this
callback with `resilient_parsing=False`), and `trader run --help`
demonstrably created `logs/trader.log`'s parent directory anyway. The
actual fix is in `configure_logging()`/`_LazyDirRotatingFileHandler`
(`trader/logging_setup.py`): nothing is written to disk until a record is
genuinely logged, so a command that only prints usage and exits creates
nothing, and this callback does not need to know that "printing usage" is
even happening.
Runs before any command's own `try`/`except TraderError`, so it needs its
own: an invalid `LOG_LEVEL` raises `ConfigError` from `configure_logging()`
itself, and without this handler that error would propagate as a raw
traceback instead of the one clean `Error: ...` line every other bad
config value in this app produces.
"""
try:
configure_logging()
except TraderError as exc:
raise _fail(str(exc)) from exc
# --- Construction seams ------------------------------------------------------
# Commands call these instead of constructing adapters inline, so tests can
# substitute fakes and never touch the network.
[docs]
def build_broker(config: EffectiveTradingConfig) -> BrokerAdapter:
"""Build the broker adapter for the resolved config."""
return AlpacaBroker.from_config(config)
[docs]
def build_provider() -> MarketDataProvider:
"""Build the market data provider."""
return YFinanceProvider()
[docs]
def build_snapshot_repository(config: EffectiveTradingConfig) -> SnapshotRepository:
"""Build the snapshot repository against the configured database."""
engine = create_db_engine(config.database_url)
return SnapshotRepository(create_session_factory(engine))
[docs]
def build_bar_repository(config: EffectiveTradingConfig) -> BarRepository:
"""Build the bar repository against the configured database."""
engine = create_db_engine(config.database_url)
return BarRepository(create_session_factory(engine))
[docs]
def build_bar_cache(
config: EffectiveTradingConfig, refresh_tail: timedelta | None = None
) -> BarCache:
"""Build the bar cache over the configured database and provider."""
return BarCache(build_bar_repository(config), build_provider(), refresh_tail)
[docs]
def build_backtest_repository(config: EffectiveTradingConfig) -> BacktestRepository:
"""Build the backtest repository against the configured database."""
engine = create_db_engine(config.database_url)
return BacktestRepository(create_session_factory(engine))
[docs]
def build_trade_repository(config: EffectiveTradingConfig) -> TradeRepository:
"""Build the trade repository against the configured database."""
engine = create_db_engine(config.database_url)
return TradeRepository(create_session_factory(engine))
[docs]
def build_decision_repository(config: EffectiveTradingConfig) -> DecisionRepository:
"""Build the decision repository against the configured database."""
engine = create_db_engine(config.database_url)
return DecisionRepository(create_session_factory(engine))
[docs]
def build_outcome_repository(config: EffectiveTradingConfig) -> OutcomeRepository:
"""Build the outcome repository against the configured database."""
engine = create_db_engine(config.database_url)
return OutcomeRepository(create_session_factory(engine))
[docs]
def build_simulation_repository(config: EffectiveTradingConfig) -> SimulationRepository:
"""Build the shadow-portfolio simulator's repository against the
configured database (issue #33)."""
engine = create_db_engine(config.database_url)
return SimulationRepository(create_session_factory(engine))
[docs]
def build_watchlist_event_repository(
config: EffectiveTradingConfig,
) -> WatchlistEventRepository:
"""Build the watchlist-event (discovery audit) repository."""
engine = create_db_engine(config.database_url)
return WatchlistEventRepository(create_session_factory(engine))
[docs]
def build_deployed_version_repository(
config: EffectiveTradingConfig,
) -> DeployedVersionRepository:
"""Build the deployed-versions repository against the configured database."""
engine = create_db_engine(config.database_url)
return DeployedVersionRepository(create_session_factory(engine))
def _record_deployed_version(config: EffectiveTradingConfig) -> None:
"""Best-effort: record this process's git version if it changed (issue #30).
Never blocks trading. A version-log failure — an unmigrated database, a
`git` binary problem — is logged and swallowed, the same discipline
discovery and the LLM path use for anything that is not itself the
trading decision.
"""
try:
git_sha, git_ref = detect_git_version()
build_deployed_version_repository(config).record_if_changed(
git_sha, git_ref, now=datetime.now(UTC)
)
except Exception: # noqa: BLE001 - a version-log failure must not block trading
_logger.warning("could not record the deployed version", exc_info=True)
[docs]
def build_index_membership_repository(
config: EffectiveTradingConfig,
) -> IndexMembershipRepository:
"""Build the index-membership repository against the configured database."""
engine = create_db_engine(config.database_url)
return IndexMembershipRepository(create_session_factory(engine))
[docs]
def build_analyst_consensus_repository(
config: EffectiveTradingConfig,
) -> AnalystConsensusRepository:
"""Build the analyst-consensus-history repository against the configured database."""
engine = create_db_engine(config.database_url)
return AnalystConsensusRepository(create_session_factory(engine))
[docs]
def build_focus_list_repository(config: EffectiveTradingConfig) -> FocusListRepository:
"""Build the `focus_list` repository against the configured database
(issue #44)."""
engine = create_db_engine(config.database_url)
return FocusListRepository(create_session_factory(engine))
[docs]
def build_news_archive_repository(
config: EffectiveTradingConfig,
) -> NewsArchiveRepository:
"""Build the news-archive repository against the configured database."""
engine = create_db_engine(config.database_url)
return NewsArchiveRepository(create_session_factory(engine))
[docs]
def build_news_archive(config: EffectiveTradingConfig) -> NewsArchive:
"""Build the news archive over the configured database and Alpaca's history.
Alpaca only, and not `build_news_provider`'s merge: yfinance has no archive
to contribute — roughly the last ten items per ticker — so a merged provider
could not answer for a past window at all. The consequence is recorded rather
than hidden: a replayed prompt is thinner than the live one for the names
whose yfinance coverage is better, which bounds the replay's conclusion to
the model's skill at reading the Alpaca corpus.
"""
return NewsArchive(
build_news_archive_repository(config),
AlpacaNewsProvider(api_key=config.api_key, api_secret=config.api_secret),
)
[docs]
def build_news_provider(
config: EffectiveTradingConfig,
*,
news_window_hours: float = DEFAULT_NEWS_WINDOW_HOURS,
) -> NewsProvider:
"""Build the news provider: yfinance and Alpaca (Benzinga), merged.
The two feeds are **disjoint corpora** — 0 of 10 shared headlines on every
symbol probed, measured 2026-08-14 — so this is additive, not a choice
between them. yfinance is the primary because it carries a summary far more
often, and the merge prefers the primary's copy of a shared story.
`news_window_hours` is the strategy's own `max_news_age_hours`, passed in so
that the merge's freshness ordering and the strategy's filter cannot mean
different things by "fresh".
"""
return MergedNewsProvider(
YFinanceNewsProvider(),
AlpacaNewsProvider(api_key=config.api_key, api_secret=config.api_secret),
freshness_hours=news_window_hours,
)
[docs]
def build_search_provider() -> NewsSearchProvider:
"""Build the phrase-search provider discovery searches each theme with."""
return YFinanceSearchProvider()
[docs]
@functools.lru_cache(maxsize=1)
def build_analyst_provider() -> AnalystProvider:
"""Build the analyst-consensus provider discovery filters candidates with.
`lru_cache(maxsize=1)` makes this a process-lifetime singleton (issue
#100): every call site gets the same instance, so
`YFinanceAnalystProvider`'s own per-symbol cache actually has repeat
calls to hit, instead of starting empty every cycle. Every test that
touches this replaces the whole name via `monkeypatch.setattr(cli,
"build_analyst_provider", ...)`, which bypasses the cache entirely — the
real, decorated function is never called during the test suite.
"""
return YFinanceAnalystProvider()
[docs]
@functools.lru_cache(maxsize=1)
def build_earnings_provider() -> EarningsProvider:
"""Build the earnings-calendar provider `LlmStrategy` reads (issue #89).
Same process-lifetime-singleton reasoning as `build_analyst_provider`
immediately above (issue #100): every call site shares one instance, so
`YFinanceEarningsProvider`'s own per-symbol cache actually accumulates
across cycles. Every test that touches this replaces the whole name via
`monkeypatch.setattr(cli, "build_earnings_provider", ...)`, the same way,
which bypasses the cache entirely.
"""
return YFinanceEarningsProvider()
[docs]
@functools.lru_cache(maxsize=1)
def build_sector_provider() -> SectorProvider:
"""Build the sector/industry provider `--risk` reports classify with.
Same singleton reasoning as `build_analyst_provider` (issue #100).
"""
return YFinanceSectorProvider()
[docs]
def build_index_provider() -> WikipediaIndexProvider:
"""Build the index-membership scraper `trader scan-universe` uses."""
return WikipediaIndexProvider()
[docs]
def build_discovery_settings() -> DiscoverySettings:
"""Read discovery settings from `config/watchlist.yaml`.
A separate seam from `load_watchlist()` (which reads the same file for
`fixed_tickers`) so a test can control discovery's themes/filters without
also having to fake the fixed-ticker list, and vice versa.
"""
return load_watchlist_config(Path("config/watchlist.yaml")).discovery
[docs]
def build_llm_provider(
model: str,
timeout_seconds: int = 120,
temperature: float = 0.0,
num_ctx: int = DEFAULT_NUM_CTX,
) -> LlmProvider:
"""Build the Ollama provider from settings and strategy parameters."""
return OllamaProvider(
model=model,
base_url=Settings().ollama_base_url,
timeout_seconds=timeout_seconds,
temperature=temperature,
num_ctx=num_ctx,
)
[docs]
def build_ranker() -> Ranker | None:
"""An `LlmRanker` on the configured model, or `None` for the fallback.
The model comes from the configured `llm` strategy's own params rather than
from a second setting: one model name in one place cannot drift out of step
with itself. A configuration with no `llm` strategy gets `None`, which
`run_once` reads as confidence order — a rule-only setup should not stand
up a provider pointed at a model nobody asked for.
Constructs only. `OllamaProvider.__init__` opens no connection, so this is
safe to call on a machine with no model server: a ranking failure then
degrades that one cycle's order rather than failing at startup.
"""
params = llm_strategy_params()
if not params:
return None
return LlmRanker(
build_llm_provider(
model=str(params.get("model") or "qwen2.5:32b"),
timeout_seconds=int(params.get("timeout_seconds") or 120),
temperature=float(params.get("temperature") or 0.0),
num_ctx=int(params.get("num_ctx") or DEFAULT_NUM_CTX),
)
)
[docs]
def build_replay_runner(config: EffectiveTradingConfig) -> ReplayRunner:
"""Build a `ReplayRunner` for issue #7 step 4, over the configured database.
The model, window and lookback come from the configured `llm` strategy's
own params — the same source `build_ranker` reads — so a replay asks the
same question production would have asked, not a second, independently
drifting copy of the same settings. `build_news_archive` (Alpaca only,
never the merged live feed) and `build_provider` (yfinance, for one
forward-return fetch per symbol) are the same seams the rest of the CLI
uses, so a test can substitute fakes for both without touching the
network.
"""
params = llm_strategy_params()
return ReplayRunner(
news_repository=build_news_archive(config),
bar_repository=build_bar_repository(config),
market_data=build_provider(),
llm_provider_factory=lambda: build_llm_provider(
model=str(params.get("model") or "qwen2.5:32b"),
timeout_seconds=int(params.get("timeout_seconds") or 120),
temperature=float(params.get("temperature") or 0.0),
num_ctx=int(params.get("num_ctx") or DEFAULT_NUM_CTX),
),
window_hours=float(params.get("max_news_age_hours") or DEFAULT_NEWS_WINDOW_HOURS),
lookback_bars=int(params.get("lookback_bars") or 30),
)
[docs]
def llm_strategy_params() -> dict[str, object]:
"""The params of the first configured `llm` strategy, or `{}`.
`llm-check` needs the model name, and the model name lives with the
strategy that uses it rather than in a second config file.
"""
for entry in load_strategies():
if entry.type == "llm":
return dict(entry.params)
return {}
[docs]
def load_strategies() -> list[StrategyConfig]:
"""Read the configured strategies from `config/strategies.yaml`."""
return load_strategies_config(Path("config/strategies.yaml")).strategies
[docs]
def build_pipeline_config() -> PipelineConfig:
"""Read `config/pipeline.yaml`."""
return load_pipeline_config(Path("config/pipeline.yaml"))
[docs]
def build_reporting_config() -> ReportingConfig:
"""Read `config/reporting.yaml`, falling back to defaults if it is absent.
Same "absence is fine, wrongness is not" rule as `build_daemon_config`: a
fresh clone with no `reporting.yaml` still gets a usable QQQ/VOO default,
but a file that exists and fails the schema is a real error.
"""
path = Path("config/reporting.yaml")
return load_reporting_config(path) if path.exists() else ReportingConfig()
[docs]
def build_ideas_scraper_config() -> IdeasScraperConfig:
"""Read `config/ideas_scraper.yaml`, falling back to defaults if it is
absent — same "absence is fine, wrongness is not" rule as
`build_reporting_config` (issue #75)."""
path = Path("config/ideas_scraper.yaml")
return load_ideas_scraper_config(path) if path.exists() else IdeasScraperConfig()
[docs]
def build_simulation_settings() -> SimulationSettings:
"""Read `config/simulation.yaml`, falling back to defaults if it is
absent — same "absence is fine, wrongness is not" rule as
`build_reporting_config` (issue #33)."""
path = Path("config/simulation.yaml")
return load_simulation_config(path) if path.exists() else SimulationSettings()
[docs]
def build_simulation_runner(
config: EffectiveTradingConfig,
pipeline_config: PipelineConfig,
discovery: DiscoverySettings,
fixed: Collection[str],
) -> SimulationRunner:
"""Bundle the shadow-portfolio simulator's collaborators (issue #33).
Reuses `config`'s own database rather than a second connection — the
simulator reads the same `decisions`/`bars` rows the real cycle just
wrote, through repositories built the same way every other one in this
module is.
"""
return SimulationRunner(
settings=build_simulation_settings(),
pipeline_config=pipeline_config,
discovery_settings=discovery,
fixed_symbols=fixed,
bar_repository=build_bar_repository(config),
decision_repository=build_decision_repository(config),
simulation_repository=build_simulation_repository(config),
)
[docs]
def build_daemon_config() -> DaemonConfig:
"""Read `config/daemon.yaml`, falling back to defaults if it is absent.
Unlike the other configs, a missing file is not an error: the daemon's
defaults are usable, and a fresh clone should be able to run without
creating a file first. A file that exists but is *wrong* is still an
error — the fallback is for absence, not for invalidity.
"""
path = Path("config/daemon.yaml")
return load_daemon_config(path) if path.exists() else DaemonConfig()
[docs]
def load_watchlist() -> list[str]:
"""The fixed tickers from `config/watchlist.yaml`."""
return load_watchlist_config(Path("config/watchlist.yaml")).fixed_tickers
[docs]
def load_benchmark_symbol() -> str:
"""The relative-strength benchmark symbol from `config/watchlist.yaml`
(issue #87). A third narrow seam alongside `load_watchlist()` and
`build_discovery_settings()`, same reasoning as both: a test controlling
one of the three should not have to also fake the other two."""
return load_watchlist_config(Path("config/watchlist.yaml")).benchmark_symbol
[docs]
def build_chat_service() -> ChatService:
"""Build the chat CLI's `ChatService` (requirements §4, §13; issues #13-16).
Deliberately built from `Settings()` alone, never `_load_config()`: chat
is a second entry point into the Core Service Layer that must work
whether or not the trading loop's broker credentials are configured —
the same reasoning `init_db_command` and `migrate` already use for the
database path in isolation. Chat reads `trades`/`decisions`/
`account_snapshots` and `config/strategies.yaml`, runs a backtest, and
talks to Ollama; none of that touches Alpaca.
A strategy config error degrades to "no strategies configured" rather
than failing chat outright — a broken `strategies.yaml` should not also
take down the one tool that can explain what is wrong with it (issue #16).
"""
_chat_settings = Settings()
engine = create_db_engine(
assemble_database_url(
_chat_settings.database_url_template, _chat_settings.database_password
)
)
session_factory = create_session_factory(engine)
try:
strategy_entries = load_strategies()
except TraderError:
strategy_entries = []
# Not `llm_strategy_params()`: that helper re-reads
# `config/strategies.yaml` itself, which would raise a second time on
# exactly the broken-config case the `try` above exists to survive.
# Derived from `strategy_entries` instead, which is already `[]` if the
# file could not be read.
params: dict[str, object] = {}
for entry in strategy_entries:
if entry.type == "llm":
params = dict(entry.params)
break
llm = build_llm_provider(
model=str(params.get("model") or "qwen2.5:32b"),
timeout_seconds=int(params.get("timeout_seconds") or 120),
temperature=float(params.get("temperature") or 0.0),
num_ctx=int(params.get("num_ctx") or DEFAULT_NUM_CTX),
)
return ChatService(
llm=llm,
decisions=DecisionRepository(session_factory),
trades=TradeRepository(session_factory),
snapshots=SnapshotRepository(session_factory),
outcomes=OutcomeRepository(session_factory),
strategy_entries=strategy_entries,
strategies_path=Path("config/strategies.yaml"),
bar_cache=BarCache(BarRepository(session_factory), build_provider()),
backtest_repository=BacktestRepository(session_factory),
)
def _build_alias_resolver(broker: BrokerAdapter) -> CompanyAliasResolver:
"""A symbol -> company-alias lookup backed by the broker's asset record.
Alpaca already sends the issuer's name with every asset, and every
evaluated symbol already goes through `get_asset` for the tradability
gate — so this adds no network call the cycle was not already making. It
feeds `relevance_count` and nothing else (issue #6): no order, no prompt
and no guardrail reads it, which is why a failure here is allowed to
degrade quietly to bare-ticker matching.
Passing the broker rather than a bound method keeps the `None` handling —
"the broker has never heard of this symbol" — in one place.
"""
def fetch_name(symbol: str) -> str | None:
asset = broker.get_asset(symbol)
return None if asset is None else asset.name
return CompanyAliasResolver(fetch_name)
[docs]
def build_strategy_runs(
entries: Sequence[StrategyConfig],
pipeline_config: PipelineConfig,
decision_memory: DecisionMemory | None = None,
*,
config: EffectiveTradingConfig,
alias_resolver: Callable[[str], Sequence[str]] | None = None,
analyst_provider: AnalystProvider | None = None,
earnings_provider: EarningsProvider | None = None,
) -> list[StrategyRun]:
"""Build every configured strategy, with its mode and its evaluator.
An `llm` entry gets a provider built from its own params; a rule entry gets
a `BacktestEvaluator`. A non-backtestable strategy gets no evaluator at all,
since one could not gate it.
`decision_memory` is what lets an `llm` entry recognise that it already
answered this question — the news-fingerprint gate. It is optional, and
defaults to `None` so that a caller with no database (the CLI's own tests,
and `llm-check`) builds a strategy that simply calls the model every cycle.
The production callers pass the same `DecisionRepository` they hand to
`run_once`, so a decision is read back through the object that wrote it.
`config` is keyword-only and **required**: the Alpaca half of the news feed
is built from its credentials, and a default would let a caller produce a
credential-less feed by omission, whose first symptom is a log line that
reads like a provider outage.
"""
runs: list[StrategyRun] = []
for entry in entries:
llm = news = None
if entry.type in ("llm", "trailing_stop_floor"):
# Both strategy types need an `LlmProvider`, built from the
# entry's own model/temperature/timeout_seconds/num_ctx params —
# see `registry._PROVIDER_PARAMS`. Only `llm` also needs a news
# feed: `trailing_stop_floor` reads a companion notes file, not
# a news provider, so `news` stays `None` for it.
llm = build_llm_provider(
model=str(entry.params.get("model") or "qwen2.5:32b"),
timeout_seconds=int(entry.params.get("timeout_seconds") or 120),
temperature=float(entry.params.get("temperature") or 0.0),
num_ctx=int(entry.params.get("num_ctx") or DEFAULT_NUM_CTX),
)
if entry.type == "llm":
window = entry.params.get("max_news_age_hours")
news = build_news_provider(
config,
news_window_hours=(
float(window) if window is not None else DEFAULT_NEWS_WINDOW_HOURS
),
)
strategy = build_strategy(
entry,
llm=llm,
news_provider=news,
decision_memory=decision_memory,
alias_resolver=alias_resolver,
analyst_provider=analyst_provider,
earnings_provider=earnings_provider,
)
evaluator = (
BacktestEvaluator(strategy, pipeline_config)
if strategy.backtestable
else None
)
runs.append(StrategyRun(strategy, entry.mode, evaluator))
return runs
def _recent_bars(cache: BarCache, symbol: str, settings: DiscoverySettings) -> list[Bar]:
"""A short recent bar window for discovery's price/volume filters.
`settings` is accepted to match the shape discovery is wired with, but is
not read: there is no trend filter (removed in Task 5), so nothing here
needs a configurable lookback. A fixed `_DISCOVERY_BAR_WINDOW_DAYS` is
comfortably enough daily bars for `min_price` (the last close) and
`min_avg_volume` (a short recent average).
"""
del settings
now = datetime.now(UTC)
return cache.ensure(symbol, now - timedelta(days=_DISCOVERY_BAR_WINDOW_DAYS), now)
def _screen_for_entry_gate(
*, discovery, broker, cache, fixed, focus_list=()
) -> dict[str, FilterOutcome]:
"""Run the operator's own tickers, plus today's focus list, through the
discovery gates.
Entry-blocking only: a failing symbol is still evaluated and still records
a decision, because `fixed_tickers` is a watchlist, not a buy list — "I
don't know that these are good purchases, I only know I am interested in
watching them". A focus-list symbol gets the same treatment (issue #44,
Decision 10): its chain result can be up to ~24 hours stale by the time an
entry is attempted, which today's live-evaluated discoveries never are, so
a same-cycle sanity check on tradability/analyst-opinion/price-volume is a
net strengthening, not a new restriction on anything that trades today.
Never raises, so a screening outage cannot stop trading. That swallowing
is also why `test_the_cli_helper_actually_produces_blocks` exists: the
first version of this passed the wrong arguments to `_recent_bars`, and
without a test the failure was silent and blocked nothing forever.
"""
symbols = list(dict.fromkeys([*fixed, *focus_list]))
if not symbols:
return {}
try:
return screen_for_entry(
symbols,
settings=discovery,
analysts=build_analyst_provider(),
broker=broker,
bars_for=lambda s: _recent_bars(cache, s, discovery),
)
except Exception: # noqa: BLE001 - screening must never break a cycle
logging.getLogger("trader.discovery").warning(
"could not screen fixed tickers and the focus list; none blocked this cycle",
exc_info=True,
)
return {}
def _discover_symbols_guarded(
*,
discovery: DiscoverySettings,
broker: BrokerAdapter,
cache: BarCache,
fixed: Collection[str],
event_repository: WatchlistEventRepository,
index_repository: IndexMembershipRepository | None = None,
consensus_repository: AnalystConsensusRepository | None = None,
) -> list[str]:
"""The symbols `discover_symbols` surfaced this cycle, or `[]`.
`trader.discovery.scan.discover_symbols` documents that it never raises —
every per-theme and per-candidate failure is already caught inside it.
This `try` is the outer belt anyway, on the same discipline the LLM path
uses to always resolve to HOLD: if discovery somehow raises past its own
guarantee, the fixed watchlist must still trade rather than the whole
cycle failing.
`index_repository`/`consensus_repository` default to `None` (issue #27):
a caller that omits them gets theme search only, unchanged — the
analyst-scan source is additive.
"""
try:
candidates = discover_symbols(
settings=discovery,
search=build_search_provider(),
analysts=build_analyst_provider(),
broker=broker,
bars_for=lambda s: _recent_bars(cache, s, discovery),
fixed_symbols=fixed,
now=datetime.now(UTC),
event_repository=event_repository,
index_repository=index_repository,
consensus_repository=consensus_repository,
)
except Exception: # noqa: BLE001 - discovery must never stop fixed tickers trading
_logger.error(
"discovery: scan raised past its own no-raise guarantee; "
"continuing this cycle with the fixed watchlist only",
exc_info=True,
)
return []
return [candidate.symbol for candidate in candidates]
def _session_date(clock: MarketClock) -> date:
"""The trading-session date intraday reads `focus_list` under (issue #44,
Decision 7).
`trader build-focus-list` stamps every row with `next_open.date()` at the
moment it runs, off market hours, in the exchange's own calendar — never
wall-clock UTC midnight (`BarCoverage` already taught this codebase what
that costs). Reading the same field back only works while the market is
closed, though: `MarketClock.next_open` is documented as "meaningful even
while the market is open — it is the *following* session", so while
trading is live it names *tomorrow*, not today. `next_close`, symmetrically,
is today's close whenever the market is open — both timestamps convert to
UTC well inside the exchange's own trading day, so `.date()` on either
never crosses a UTC calendar boundary. Branching on `is_open` (the same
signal `MarketSchedule` branches on) picks whichever field names *today*:
`next_close` while trading, `next_open` while closed — matching exactly
what the batch itself stamped when it ran the night before.
"""
return clock.next_close.date() if clock.is_open else clock.next_open.date()
def _todays_focus_list(
broker: BrokerAdapter, repository: FocusListRepository
) -> list[str]:
"""Today's `trader build-focus-list` output, or `[]` (issue #44).
Never raises: a clock or database read failure here must not cost the
cycle, the same discipline `_discover_symbols_guarded` and
`_screen_for_entry_gate` already use. A read failure degrades exactly like
a batch that never ran — `focus_list` contributes nothing this cycle,
fixed tickers and theme-search discoveries still trade.
"""
try:
trading_day = _session_date(broker.get_clock())
return repository.symbols_for(trading_day)
except Exception: # noqa: BLE001 - a focus-list read failure must not cost the cycle
_logger.warning(
"could not read today's focus list; none folded into this cycle",
exc_info=True,
)
return []
def _build_universe(
fixed: Collection[str],
discovered: Collection[str],
focus_list: Collection[str],
held: Collection[str],
) -> list[str]:
"""Fixed tickers first, then this cycle's theme-search discoveries, then
today's focus list, then anything held (issue #44, Decision 8).
Fixed-first so the operator's own tickers are evaluated before any
discovery when a cycle is truncated by anything. `dict.fromkeys`
deduplicates while preserving that order — a symbol appearing in more than
one group (already fixed and discovered, already fixed and on the focus
list, or any of those and already held) must not be evaluated twice.
`focus_list` is a precomputed result from last night's
`trader build-focus-list`, not a candidate source re-evaluated here —
re-running the filter chain intraday on a focus-list symbol would
reintroduce the live analyst-scan cost this design removes.
`held` runs last and unconditionally: a position holding a symbol whose
story has faded out of the scan (or dropped off the next night's focus
list) must still reach `run_once`, or the app can never sell what it
bought. That is the one bug in this design that would cost real money.
"""
return list(dict.fromkeys([*fixed, *discovered, *focus_list, *held]))
[docs]
def execute_cycle(
*,
config: EffectiveTradingConfig,
pipeline_config: PipelineConfig,
strategy_entries: Sequence[StrategyConfig],
symbols: list[str],
dry_run: bool,
) -> CycleReport:
"""Wire the collaborators and run one cycle.
A seam: the CLI test asserts what the command *asked for* without standing
up the whole pipeline, and `run_once` stays free of construction concerns.
`symbols` is the operator-curated, fixed list from `config/watchlist.yaml`
(`fixed_tickers`) — the same meaning it had before discovery existed. A
fresh discovery scan runs every call (this command is one cycle per
process invocation), and a symbol currently held is folded into the
universe regardless of whether it is still discovered.
"""
broker = build_broker(config)
cache = build_bar_cache(
config, refresh_tail=timedelta(days=pipeline_config.bar_refresh_days)
)
discovery = build_discovery_settings()
fixed = symbols
discovered = _discover_symbols_guarded(
discovery=discovery,
broker=broker,
cache=cache,
fixed=fixed,
event_repository=build_watchlist_event_repository(config),
)
held = [p.symbol for p in broker.get_positions()]
focus_list = _todays_focus_list(broker, build_focus_list_repository(config))
universe = _build_universe(fixed, discovered, focus_list, held)
entry_blocked = _screen_for_entry_gate(
discovery=discovery,
broker=broker,
cache=cache,
fixed=fixed,
focus_list=focus_list,
)
# One repository object, wired twice on purpose: `run_once` writes decisions
# through it, and an `llm` strategy reads its own previous decision back
# through the same object. Two instances would work — they share a database
# — but one makes it obvious that the thing being read is the thing that was
# written, which is the whole basis of the news-fingerprint gate.
decision_repository = build_decision_repository(config)
return run_once(
broker=broker,
cache=cache,
strategies=build_strategy_runs(
strategy_entries,
pipeline_config,
decision_repository,
config=config,
alias_resolver=_build_alias_resolver(broker),
analyst_provider=build_analyst_provider(),
earnings_provider=build_earnings_provider(),
),
guardrails=Guardrails(pipeline_config),
executor=OrderExecutor(broker, pipeline_config, dry_run=dry_run),
trade_repository=build_trade_repository(config),
decision_repository=decision_repository,
snapshot_repository=build_snapshot_repository(config),
outcome_repository=build_outcome_repository(config),
symbols=universe,
fixed_symbols=fixed,
max_discovered_positions=discovery.max_discovered_positions,
entry_blocked=entry_blocked,
pipeline_config=pipeline_config,
ranker=build_ranker(),
simulation=build_simulation_runner(config, pipeline_config, discovery, fixed),
benchmark_symbol=load_benchmark_symbol(),
)
[docs]
def build_daemon_cycle(
*,
config: EffectiveTradingConfig,
pipeline_config: PipelineConfig,
daemon_config: DaemonConfig,
strategy_entries: Sequence[StrategyConfig],
symbols: list[str],
dry_run: bool,
) -> tuple[Callable[[dict[str, Position] | None], CycleReport], BrokerAdapter]:
"""Build the daemon's object graph once, and bind it into a cycle callable.
Built once for the process's life, deliberately: config edits take effect
on restart, and flipping a strategy to `mode: trading` should happen at a
moment the operator chose rather than at whichever tick follows a file
save.
Returns the cycle callable and a *read-only* broker for the loop to check
the clock and snapshot positions with. Two different objects over one
connection on purpose — the loop cannot place an order because the thing
it holds has no method to do so, while `run_once` keeps the raw
order-placing broker it needs. Discovery's own `get_asset` read goes
through the same read-only broker, for the same reason `get_clock` does.
`symbols` is the operator-curated, fixed list (`fixed_tickers`) — its
meaning is unchanged from before discovery existed. Discovery's settings
are read once here, at the same "built once for the process's life" point
as everything else in this function; only the *scan itself* — the part
that changes cycle to cycle as new news arrives — runs inside `cycle`,
which is what makes it recompute every cycle rather than once at startup.
"""
broker = build_broker(config)
cache = BarCache(
build_bar_repository(config),
RetryingMarketData(
build_provider(),
attempts=daemon_config.retry_attempts,
base_seconds=daemon_config.retry_base_seconds,
),
refresh_tail=timedelta(days=pipeline_config.bar_refresh_days),
)
# Built before the strategies, because an `llm` strategy needs it: the
# news-fingerprint gate reads the previous cycle's decision back through the
# same repository that wrote it.
decision_repository = build_decision_repository(config)
# Built before the strategies too, and for the same reason discovery uses
# it: the alias lookup is a *read*, so it goes through the retrying
# read-only wrapper rather than the order-placing broker. Moving this above
# `build_strategy_runs` is the whole reason it is constructed here rather
# than just below.
read_only_broker = RetryingBroker(
broker,
attempts=daemon_config.retry_attempts,
base_seconds=daemon_config.retry_base_seconds,
)
strategies = build_strategy_runs(
strategy_entries,
pipeline_config,
decision_repository,
config=config,
alias_resolver=_build_alias_resolver(read_only_broker),
analyst_provider=build_analyst_provider(),
earnings_provider=build_earnings_provider(),
)
# Built here, with the rest of the graph, and not inside `cycle`: the
# ranker is policy read from config, and config edits take effect on
# restart at a moment the operator chose.
ranker = build_ranker()
guardrails = Guardrails(pipeline_config)
executor = OrderExecutor(broker, pipeline_config, dry_run=dry_run)
trade_repository = build_trade_repository(config)
snapshot_repository = build_snapshot_repository(config)
outcome_repository = build_outcome_repository(config)
event_repository = build_watchlist_event_repository(config)
focus_list_repository = build_focus_list_repository(config)
discovery = build_discovery_settings()
fixed = symbols
simulation = build_simulation_runner(config, pipeline_config, discovery, fixed)
# Same "built once, restart to pick up a config edit" discipline as
# `ranker` above (issue #87).
benchmark_symbol = load_benchmark_symbol()
def cycle(previous_positions: dict[str, Position] | None) -> CycleReport:
# `index_repository`/`consensus_repository` are no longer passed here
# (issue #44, Decision 8): the analyst-scan candidate source moved to
# the overnight `trader build-focus-list` batch, so the live cycle's
# own `discover_symbols` call runs theme search only, at the same cost
# it had before issue #27 — the literal live-scale trigger this design
# removes.
discovered = _discover_symbols_guarded(
discovery=discovery,
broker=read_only_broker,
cache=cache,
fixed=fixed,
event_repository=event_repository,
)
held = [p.symbol for p in read_only_broker.get_positions()]
focus_list = _todays_focus_list(read_only_broker, focus_list_repository)
universe = _build_universe(fixed, discovered, focus_list, held)
entry_blocked = _screen_for_entry_gate(
discovery=discovery,
broker=read_only_broker,
cache=cache,
fixed=fixed,
focus_list=focus_list,
)
return run_once(
broker=broker,
cache=cache,
strategies=strategies,
guardrails=guardrails,
executor=executor,
trade_repository=trade_repository,
decision_repository=decision_repository,
snapshot_repository=snapshot_repository,
outcome_repository=outcome_repository,
symbols=universe,
pipeline_config=pipeline_config,
previous_positions=previous_positions,
fixed_symbols=fixed,
max_discovered_positions=discovery.max_discovered_positions,
entry_blocked=entry_blocked,
ranker=ranker,
simulation=simulation,
benchmark_symbol=benchmark_symbol,
)
return cycle, read_only_broker
def _database_reason(exc: Exception) -> str:
"""The readable part of a database failure, for a one-line CLI error.
SQLAlchemy's `str()` appends the offending SQL, its bound parameters and a
`sqlalche.me` link. That belongs in a traceback, not in a message whose
whole job is to say "run 'trader migrate'"; `exc.orig` is the driver's own
error, which is the sentence a user can act on.
"""
original = getattr(exc, "orig", None)
return f"{type(exc).__name__}: {original if original is not None else exc}"
def _fail(message: str) -> typer.Exit:
"""Print an error and return an exit(1) for the caller to raise."""
typer.echo(f"Error: {message}")
return typer.Exit(code=1)
def _load_config() -> EffectiveTradingConfig:
"""Resolve config, converting our errors into a clean CLI failure."""
try:
return load_trading_config()
except TraderError as exc:
raise _fail(str(exc)) from exc
# --- Reporting helpers (shared by `positions`, `account`, `outcomes`) --------
def _fmt_pct(value: Decimal | None) -> str:
"""A signed percentage, or `MISSING` for a value that could not be computed."""
return f"{value:+}%" if value is not None else MISSING
def _fmt_days(value: int | None) -> str:
"""A whole day count, or `MISSING` for an unknown holding period."""
return str(value) if value is not None else MISSING
def _open_round_trips(
outcome_repo: OutcomeRepository, positions: Sequence[Position]
) -> dict[str, RoundTrip | None]:
"""The open round trip behind each held symbol, best-effort.
`positions` and `account` have never required a database — a fresh
install can run `trader positions` against nothing but the broker.
Adding "days held" must not turn that into a hard dependency: a symbol
with no local round trip (an unsynced holding reconciled from a second
machine, per issue #8) or a lookup that fails outright (no database yet,
a schema older than `round_trips`) both fall back to unknown here, so the
listing itself never fails over a column that used to not exist.
"""
trips: dict[str, RoundTrip | None] = {}
for position in positions:
try:
trips[position.symbol] = outcome_repo.open_round_trip_for(position.symbol)
except Exception: # noqa: BLE001 - degrade to unknown, never break a read-only listing
trips[position.symbol] = None
return trips
def _positions_table(
positions: Sequence[Position],
open_trips: dict[str, RoundTrip | None],
now: datetime,
) -> PrettyTable:
"""The open-positions table shared by `positions` and `account`.
RETURN % calls `position_return_pct` (`trader/reporting.py`) — the same
function the LLM prompt uses to describe a held position — so the
operator reading this table and the model reading the prompt never see
two different numbers for the same position. DAYS HELD and
AVG %/DAY (SIMPLE) come from the position's open round trip in
`round_trips`; a position with none on record (see `_open_round_trips`)
renders both as a dash, never a fabricated `0`.
The TOTAL row (issue #91): QTY, PRICE, and VALUE are literal sums across
open positions. ENTRY is `total_basis` (summed cost basis), not a summed
per-share price — a literal sum there would mix entry prices of
differently-priced symbols into a meaningless number. DAYS HELD is the
average holding period; AVG %/DAY (SIMPLE) is the simple average of each
position's own average — both computed only over positions with a known
round trip, so one unsynced holding (`_open_round_trips`) doesn't pull
either average toward a fabricated value.
"""
headers = [
"SYMBOL",
"QTY",
"ENTRY",
"PRICE",
"VALUE",
"P/L",
"RETURN %",
"DAYS HELD",
"AVG %/DAY (SIMPLE)",
]
rows: list[list[object]] = []
total_pl = Decimal(0)
total_basis = Decimal(0)
total_qty = Decimal(0)
total_price = Decimal(0)
total_value = Decimal(0)
held_days: list[int] = []
avg_daily_pcts: list[Decimal] = []
for position in positions:
pct = position_return_pct(position)
trip = open_trips.get(position.symbol)
held = days_held(trip.opened_at, now) if trip is not None else None
avg = (
avg_daily_return_pct(pct, held)
if pct is not None and held is not None
else None
)
rows.append(
[
position.symbol,
position.quantity,
position.avg_entry_price,
position.current_price,
position.market_value,
position.unrealized_pl,
_fmt_pct(pct),
_fmt_days(held),
_fmt_pct(avg),
]
)
total_pl += position.unrealized_pl
total_basis += cost_basis(position.avg_entry_price, position.quantity)
total_qty += position.quantity
total_price += position.current_price
total_value += position.market_value
if held is not None:
held_days.append(held)
if avg is not None:
avg_daily_pcts.append(avg)
avg_held = round(sum(held_days) / len(held_days)) if held_days else None
avg_daily_pct = (
(sum(avg_daily_pcts, Decimal(0)) / len(avg_daily_pcts)).quantize(PERCENT_QUANTUM)
if avg_daily_pcts
else None
)
table = build_table(headers, rows)
table.add_row(
[
"TOTAL",
total_qty,
total_basis,
total_price,
total_value,
total_pl,
_fmt_pct(return_pct(total_pl, total_basis)),
_fmt_days(avg_held),
_fmt_pct(avg_daily_pct),
]
)
return table
def _sector_exposure_table(rows: Sequence[SectorExposure]) -> PrettyTable:
"""`--risk`'s held-position sector breakdown, dollar-weighted."""
headers = ["SECTOR", "MARKET VALUE", "% OF BOOK", "SYMBOLS"]
return build_table(
headers,
[
[row.sector, row.market_value, f"{row.pct_of_total}%", ", ".join(row.symbols)]
for row in rows
],
)
def _sector_count_table(rows: Sequence[SectorCount]) -> PrettyTable:
"""`--risk`'s discovered-candidate sector breakdown, one symbol one vote."""
headers = ["SECTOR", "COUNT", "% OF CANDIDATES", "SYMBOLS"]
return build_table(
headers,
[
[row.sector, row.count, f"{row.pct_of_total}%", ", ".join(row.symbols)]
for row in rows
],
)
def _correlation_table(correlation: PortfolioCorrelation) -> PrettyTable:
"""`--risk`'s pairwise return-correlation table across held positions."""
headers = ["SYMBOL A", "SYMBOL B", "CORRELATION"]
return build_table(
headers,
[
[pair.symbol_a, pair.symbol_b, f"{pair.correlation:+.2f}"]
for pair in correlation.pairwise
],
)
def _print_risk_report(
config: EffectiveTradingConfig, positions: Sequence[Position], now: datetime
) -> None:
"""`--risk`'s three sections: held-sector, discovered-sector, correlation.
Issue #65 — measurement only, never a gate. Each section is independently
best-effort: a yfinance outage on the sector lookups must not hide the
correlation numbers, and vice versa, the same isolation discipline
`_open_round_trips` already uses for "days held". Nothing here places,
cancels, or modifies an order.
"""
typer.echo("\nSector concentration (held positions, by market value):")
try:
exposure = sector_exposure(positions, build_sector_provider())
except Exception: # noqa: BLE001 - a risk report must not crash the base command
_logger.warning("sector exposure computation failed", exc_info=True)
typer.echo(" Could not be computed this run — see the log.")
else:
typer.echo(
_sector_exposure_table(exposure) if exposure else " No held positions."
)
typer.echo(
f"\nSector distribution of symbols discovered in the last "
f"{_RISK_DISCOVERED_LOOKBACK_DAYS} days (one symbol, one vote):"
)
try:
cutoff = now - timedelta(days=_RISK_DISCOVERED_LOOKBACK_DAYS)
discovered = build_watchlist_event_repository(config).discovered_symbols_since(
cutoff
)
counts = sector_distribution_by_count(discovered, build_sector_provider())
except Exception: # noqa: BLE001 - a risk report must not crash the base command
_logger.warning("discovered-sector distribution failed", exc_info=True)
typer.echo(" Could not be computed this run — see the log.")
else:
typer.echo(
_sector_count_table(counts)
if counts
else " No recent discoveries on record."
)
typer.echo(
f"\nReturn correlation across held positions "
f"(last {DEFAULT_CORRELATION_LOOKBACK_DAYS} days):"
)
try:
correlation = compute_correlations(positions, build_bar_cache(config), now=now)
except Exception: # noqa: BLE001 - a risk report must not crash the base command
_logger.warning("correlation computation failed", exc_info=True)
typer.echo(" Could not be computed this run — see the log.")
else:
if correlation.average_pairwise is None:
typer.echo(
" Not enough symbols or history to compute a correlation this run."
)
else:
typer.echo(_correlation_table(correlation))
typer.echo(
f" Average pairwise correlation: {correlation.average_pairwise:+.2f}"
)
# --- Commands ----------------------------------------------------------------
[docs]
@app.command()
def version() -> None:
"""Print the installed version."""
typer.echo(package_version("claude-alpaca-trader"))
[docs]
@setup_app.command("init-db")
def init_db_command() -> None:
"""Create every table in the configured Postgres database.
Lives under `trader setup` (issue #90), not at the top level, so it reads
as deliberately administrative rather than routine in `--help` output.
"""
# Only the database connection is needed. Deliberately does NOT resolve
# the full trading config: creating tables must not require broker
# credentials.
settings = Settings()
database_url = assemble_database_url(
settings.database_url_template, settings.database_password
)
try:
init_db(database_url)
except Exception as exc: # noqa: BLE001 - surface any DB failure readably
raise _fail(
f"Could not initialize the database at "
f"{_masked_database_target(settings)}: {exc}"
) from exc
typer.echo(f"Initialized database at {_masked_database_target(settings)}")
[docs]
@app.command()
def migrate() -> None:
"""Apply any pending database migrations."""
# Only the database connection is needed, like `init-db`: migrating must
# not require broker credentials.
settings = Settings()
database_url = assemble_database_url(
settings.database_url_template, settings.database_password
)
target = _masked_database_target(settings)
before = current_revision(database_url)
try:
upgrade_to_head(database_url)
except Exception as exc: # noqa: BLE001 - surface any migration failure readably
raise _fail(
f"Migration failed for {target} ({type(exc).__name__}: {exc}). "
"The database is unchanged if the migration ran in a transaction; "
"check 'pdm run alembic current' before retrying."
) from exc
after = current_revision(database_url)
if before == after:
typer.echo(f"Already up to date at {after} ({target}).")
else:
typer.echo(f"Migrated {target}: {before} -> {after}.")
def _masked_url(url: str) -> str:
"""A DSN with its password replaced by `***`, never the real value
(issue #26) — the same masking discipline the Alpaca keys get via
`SecretStr.__repr__`, applied here since an assembled connection string
is a plain `str` once built and cannot mask itself. Every place a
database connection string reaches `typer.echo`/an f-string must go
through this first."""
return make_url(url).render_as_string(hide_password=True)
def _masked_database_target(settings: Settings) -> str:
"""The configured database target, password masked (issue #26) — built
from the password-less template alone, so it never even touches the
real secret."""
return _masked_url(settings.database_url_template)
[docs]
@app.command("config-check")
def config_check() -> None:
"""Validate configuration and report the resolved trading mode."""
settings = Settings()
# Report from the raw settings first. A diagnostic command must still say
# what it sees when validation fails — that is the situation you most need
# a diagnostic for.
typer.echo(f"trading_mode: {settings.trading_mode.value}")
typer.echo(f"live_trading_enabled: {str(settings.live_trading_enabled).lower()}")
typer.echo(f"database: {_masked_database_target(settings)}")
try:
config = resolve_trading_config(settings)
except TraderError as exc:
typer.echo(f"\nConfiguration is NOT usable: {exc}")
raise typer.Exit(code=1) from exc
typer.echo(
f"alpaca endpoint: {'paper' if config.use_paper_endpoint else 'LIVE'}"
)
if config.is_live:
typer.echo("WARNING: live mode is fully enabled. Real money is at risk.")
else:
typer.echo("Live orders are blocked. This is paper trading.")
[docs]
@app.command("llm-check")
def llm_check() -> None:
"""Report whether the local model is reachable and pulled."""
# Without logging configured, OllamaProvider's logger has no handler
# attached and falls through to Python's stderr "lastResort" handler, so
# a connection failure prints twice: once raw from the logger, once as
# the polished "NOT usable" line below. Routing it to the file instead
# leaves one clean sentence on the terminal, which is the whole point of
# a diagnostic command.
settings = Settings()
params = llm_strategy_params()
model = str(params.get("model") or "qwen2.5:32b")
timeout = int(params.get("timeout_seconds") or 120)
# Report what it is about to try before trying it, like `config-check`:
# when the call fails, the endpoint is the first thing you want to see.
typer.echo(f"endpoint: {settings.ollama_base_url}")
typer.echo(f"model: {model}")
try:
name = build_llm_provider(model=model, timeout_seconds=timeout).health()
except TraderError as exc:
typer.echo(f"\nNOT usable: {exc}")
raise typer.Exit(code=1) from exc
typer.echo(f"\nOK — {name} answered.")
[docs]
@app.command("scrape-ideas")
def scrape_ideas_command(
url: str = typer.Argument(
..., help="Page URL to fetch and mine for product-feature ideas."
),
) -> None:
"""Fetch a web page and ask a local model what claude-alpaca-trader could
build based on it. Writes a markdown file under the configured output
directory (issue #75, stage 1 — filing the resulting ideas as GitLab
issues is a separate, future, human/Claude-driven step).
Not a trading capability: touches no Position, Order, symbol, or the
`trader run`/`run-once` cycle loop.
"""
config = build_ideas_scraper_config()
llm = build_llm_provider(
model=config.model,
timeout_seconds=config.llm_timeout_seconds,
num_ctx=config.num_ctx,
)
# Report what it is about to try before trying it, like `llm-check`: when
# a multi-minute call fails, the model and URL are the first things you
# want to see.
typer.echo(f"url: {url}")
typer.echo(f"model: {config.model}")
try:
path = scrape_ideas(
url,
output_dir=Path(config.output_dir),
fetch_timeout_seconds=config.fetch_timeout_seconds,
max_page_chars=config.max_page_chars,
llm=llm,
)
except TraderError as exc:
raise _fail(str(exc)) from exc
typer.echo(f"\nWrote {path}")
[docs]
@app.command()
def account(
risk: bool = typer.Option(
False,
"--risk",
help=(
"Also print sector concentration and return-correlation across "
"held positions, plus discovered-candidate sector distribution "
"(issue #65). Measurement only, never a gate. Costs extra "
"yfinance calls (sector lookups, and bar fetches for any symbol "
"not already cached), so it is opt-in rather than printed by "
"default."
),
),
) -> None:
"""Show the account summary and open positions, and save a snapshot."""
config = _load_config()
broker = build_broker(config)
try:
summary = broker.get_account()
positions = broker.get_positions()
except TraderError as exc:
raise _fail(str(exc)) from exc
label = "paper" if summary.is_paper else "LIVE"
typer.echo(f"Account {summary.account_id} ({label})")
typer.echo(f" portfolio value: {summary.portfolio_value}")
typer.echo(f" equity: {summary.equity}")
typer.echo(f" cash: {summary.cash}")
typer.echo(f" buying power: {summary.buying_power}")
if positions:
typer.echo("\nOpen positions:")
# Best-effort: `account` has never required a database before, and
# must not start failing over a "days held" column. See
# `_open_round_trips`.
open_trips = _open_round_trips(build_outcome_repository(config), positions)
typer.echo(_positions_table(positions, open_trips, datetime.now(UTC)))
else:
typer.echo("\nNo open positions.")
if risk:
_print_risk_report(config, positions, datetime.now(UTC))
try:
repository = build_snapshot_repository(config)
snapshot_id = repository.save_snapshot(summary, positions)
except Exception as exc: # noqa: BLE001 - report any persistence failure
raise _fail(
f"Could not save snapshot ({type(exc).__name__}: {exc}). "
"If the database has not been created yet, run 'trader setup init-db'."
) from exc
typer.echo(f"\nSaved snapshot #{snapshot_id}.")
[docs]
@app.command()
def chart(
days: int = typer.Option(
30, help="Only include snapshots from the last N days. 0 or less means all time."
),
width: int = typer.Option(60, help="Sparkline width, in characters."),
) -> None:
"""Print an ASCII equity-curve sparkline from saved account snapshots (§12.3).
Reads `account_snapshots` — the rows `trader account` and `trader run(-once)`
already save every cycle — and does not talk to the broker or place any
order. Empty or single-snapshot history prints a short explanation instead
of a curve; it is not an error.
"""
config = _load_config()
repository = build_snapshot_repository(config)
start = datetime.now(UTC) - timedelta(days=days) if days > 0 else None
try:
snapshots = repository.snapshots_since(start)
except Exception as exc: # noqa: BLE001 - report any persistence failure
raise _fail(
f"Could not read snapshots ({type(exc).__name__}: {exc}). "
"If the database has not been created yet, run 'trader setup init-db'."
) from exc
window = f"last {days} days" if days > 0 else "all time"
typer.echo(f"Equity curve ({window})")
typer.echo(format_report(equity_curve_points(snapshots), width=width))
[docs]
@app.command()
def web(
host: str | None = typer.Option(None, help="Overrides WEB_HOST from .env."),
port: int | None = typer.Option(None, help="Overrides WEB_PORT from .env."),
reload: bool = typer.Option(
False,
"--reload",
help="Auto-restart on code changes (development only; uvicorn's own reload).",
),
) -> None:
"""Serve interactive equity-curve and P&L charts at http://HOST:PORT (§12.3, issue #20).
A second, browser-based renderer alongside `trader chart`/`trader
outcomes` — same underlying data, not a replacement for either. Read-only:
no route submits an order or writes to the database. Built from
`Settings()` alone, like `build_chat_service`, so it works whether or not
Alpaca credentials are configured; the only network call any page makes is
the benchmark price fetch via yfinance, same as `trader performance`.
"""
settings = Settings()
bind_host = host or settings.web_host
bind_port = port or settings.web_port
typer.echo(f"Serving at http://{bind_host}:{bind_port} — Ctrl+C to stop.")
if reload:
# uvicorn's reload mode re-imports the app fresh in a subprocess on
# every file change, which needs an import STRING, not the app object
# built below — `trader.reporting.web.asgi` is that target, built by
# the same `build_app()` so the two paths never drift.
uvicorn.run(
"trader.reporting.web.asgi:app", host=bind_host, port=bind_port, reload=True
)
return
uvicorn.run(build_app(), host=bind_host, port=bind_port)
[docs]
@app.command()
def quote(symbol: str) -> None:
"""Print the latest price for SYMBOL."""
provider = build_provider()
try:
result = provider.get_quote(symbol)
except TraderError as exc:
raise _fail(str(exc)) from exc
as_of = result.as_of.astimezone(UTC).isoformat(timespec="seconds")
typer.echo(f"{result.symbol} {result.price} (as of {as_of})")
[docs]
@app.command()
def history(
symbol: str,
start: datetime = typer.Option(..., formats=["%Y-%m-%d"], help="Start date."),
end: datetime = typer.Option(..., formats=["%Y-%m-%d"], help="End date."),
interval: str = typer.Option("1d", help="Bar interval, e.g. 1d, 1h, 15m."),
) -> None:
"""Print historical OHLCV bars for SYMBOL."""
provider = build_provider()
try:
bars = provider.get_history(symbol, start.date(), end.date(), interval)
except TraderError as exc:
raise _fail(str(exc)) from exc
headers = ["DATE", "OPEN", "HIGH", "LOW", "CLOSE", "VOLUME"]
rows = [
[
bar.timestamp.date().isoformat(),
bar.open,
bar.high,
bar.low,
bar.close,
bar.volume,
]
for bar in bars
]
typer.echo(build_table(headers, rows))
typer.echo(f"\n{len(bars)} bars for {symbol.upper()}.")
[docs]
@app.command("fetch-bars")
def fetch_bars(
symbol: str,
start: datetime = typer.Option(..., formats=["%Y-%m-%d"], help="First date."),
end: datetime | None = typer.Option(
None, formats=["%Y-%m-%d"], help="Last date; defaults to today."
),
refresh: bool = typer.Option(False, "--refresh", help="Re-fetch even if covered."),
) -> None:
"""Cache historical bars locally so backtests run offline and repeatably."""
config = _load_config()
window_end = end or datetime.now(UTC)
try:
# No refresh_tail: this command inspects an explicit historical window,
# so re-fetching its tail on every invocation would cost a request and
# change nothing. The trading paths set it; see build_bar_cache.
bars = build_bar_cache(config).ensure(
symbol.upper(),
start.replace(tzinfo=UTC),
window_end.replace(tzinfo=UTC),
refresh=refresh,
)
except TraderError as exc:
raise _fail(str(exc)) from exc
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not read or write cached bars ({type(exc).__name__}: {exc}). "
"If the database is missing or predates this version, run "
"'trader setup init-db'."
) from exc
typer.echo(
f"{symbol.upper()}: {len(bars)} bars available in {_masked_url(config.database_url)}"
)
[docs]
@app.command("seed-bars")
def seed_bars(
file: Path = typer.Option(
..., "--file", help="The Feather or IB CSV file to import."
),
symbol: str = typer.Option(..., "--symbol", help="Ticker the file covers."),
interval: str = typer.Option(
..., "--interval", help="Bar width to store under, e.g. '1d' or '1m'."
),
file_format: SeedFileFormat = typer.Option(
..., "--format", help="'feather' (yfinance-sourced) or 'ibkr-csv'."
),
tz: str = typer.Option(
"America/New_York",
"--tz",
help="Exchange timezone for --format ibkr-csv's naive timestamps. "
"Ignored for --format feather, whose timestamps are already tz-aware "
"or are yfinance's own naive-UTC convention.",
),
) -> None:
"""Seed historical bars from a third-party file into the local cache (issue #41).
A one-off import, not a live data source: writes through
`BarRepository.save_bars` — the same path `YFinanceProvider`/`BarCache`
use — so an imported bar is stored exactly like a live-fetched one,
except for `CachedBar.source`, which this command always sets to a named
vendor tag (never `None`, which is reserved for a bar this app fetched
itself). `save_bars` already skips any timestamp already stored, so
re-running this command against the same file is a safe no-op rather
than a duplicate import; `BarRepository.delete_bars` separately refuses
to delete a seeded row at all, so a later live refresh can never
silently replace validated seed data with a differently-adjusted fetch.
Only what is explicitly named on the command line is imported — no
directory scan, no "import everything under a data folder" default, per
issue #41's "import only what's actually going to be used, not all
264MB reflexively."
"""
config = _load_config()
ticker = symbol.strip().upper()
source = (
SOURCE_YFINANCE_SEED
if file_format is SeedFileFormat.FEATHER
else SOURCE_IBKR_SEED
)
try:
bars = (
load_feather_bars(file, ticker)
if file_format is SeedFileFormat.FEATHER
else load_ibkr_csv_bars(file, ticker, tz=tz)
)
except TraderError as exc:
raise _fail(str(exc)) from exc
except Exception as exc: # noqa: BLE001 - pandas/pyarrow raise many types
raise _fail(
f"{ticker}: could not parse {file} ({type(exc).__name__}: {exc})"
) from exc
if not bars:
raise _fail(f"{ticker}: {file} contained no usable rows; nothing imported.")
try:
inserted = build_bar_repository(config).save_bars(
ticker, interval, bars, source=source
)
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not write cached bars ({type(exc).__name__}: {exc}). "
"If the database is missing or predates this version, run "
"'trader setup init-db'."
) from exc
skipped = len(bars) - inserted
typer.echo(
f"{ticker}: {inserted} bar(s) imported from {file} ({file_format.value}, "
f"source={source}) into {_masked_url(config.database_url)}"
+ (f"; {skipped} already present and skipped" if skipped else "")
)
[docs]
@app.command("backfill-news")
def backfill_news(
start: datetime = typer.Option(
..., formats=["%Y-%m-%d"], help="First publication date to archive."
),
end: datetime | None = typer.Option(
None, formats=["%Y-%m-%d"], help="Exclusive upper bound; defaults to now."
),
symbol: list[str] = typer.Option(
None, "--symbol", help="Repeatable. Defaults to config/watchlist.yaml."
),
) -> None:
"""Archive Alpaca's news history locally, so past decisions can be replayed.
Reads no broker and places no orders. Alpaca's archive reaches back to 2023,
which is the property yfinance lacks and the reason this table exists.
Every symbol is attempted even if one fails, because progress is per symbol
and re-fetching what already succeeded costs requests for nothing. A failure
still exits non-zero: the archive's own coverage table is what makes a hole
detectable later, and an exit code of 0 would hide it at the one moment
someone is actually watching.
"""
config = _load_config()
window_start = start.replace(tzinfo=UTC)
window_end = (end.replace(tzinfo=UTC)) if end is not None else datetime.now(UTC)
if window_start >= window_end:
raise _fail(
f"--start ({window_start.date()}) must be before --end "
f"({window_end.date()}); the window is half-open, [start, end)."
)
try:
symbols = [s.upper() for s in symbol] if symbol else load_watchlist()
except TraderError as exc:
raise _fail(str(exc)) from exc
if not symbols:
raise _fail("No symbols to archive: pass --symbol or populate fixed_tickers.")
archive = build_news_archive(config)
failures: list[str] = []
for ticker in symbols:
try:
items = archive.ensure(ticker, window_start, window_end)
except TraderError as exc:
failures.append(ticker)
typer.echo(f"{ticker:<8}FAILED {exc}")
continue
# "in the archive", not "archived": `ensure` returns what the window
# holds, and a re-run that fetched nothing because the span was already
# covered would otherwise print the same sentence as the run that did
# the work. Observed on the live probe — SPY printed "62 items archived"
# twice, having fetched them once.
typer.echo(f"{ticker:<8}{len(items)} items in the archive")
typer.echo(
f"{len(symbols) - len(failures)} of {len(symbols)} symbols archived from "
f"{window_start.date()} to {window_end.date()} into {_masked_url(config.database_url)}"
)
if failures:
raise _fail(
f"{len(failures)} symbol(s) left incomplete: {', '.join(failures)}. "
"Re-run to fill the gap; coverage was not recorded for them."
)
#: A politeness delay between per-symbol yfinance fetches — decision #9's
#: "sequential with backoff", kept to a minimal fixed delay rather than a full
#: exponential-backoff framework, since this is the first place in the
#: codebase hitting yfinance at index scale and nothing here has been
#: rate-limit-tested at that volume yet.
_SCAN_UNIVERSE_DELAY_SECONDS = 0.2
[docs]
@app.command("scan-universe")
def scan_universe(
index: str = typer.Option(
ANALYST_SCAN_INDEX, help="Which index to scan. Only 'sp500' is supported today."
),
) -> None:
"""Scrape index membership and refresh stale analyst consensus (issue #27).
Separate from `trader run`'s daemon loop on purpose: this is the
expensive, index-scale work (~500 symbols), meant to run on its own
cadence (cron, or by hand) rather than every trading cycle.
`discover_symbols()` only ever reads what this command already wrote — a
cheap DB query, never a yfinance call at cycle scale — so a stale or
never-run scan degrades to "this source contributes nothing," the same
shape `themes: []` degrades to.
Every member is attempted even if one fails, same discipline as
`backfill-news`: a failure still exits non-zero so a real gap is never
silently hidden behind an exit code of 0.
"""
config = _load_config()
discovery = build_discovery_settings()
index_repository = build_index_membership_repository(config)
consensus_repository = build_analyst_consensus_repository(config)
now = datetime.now(UTC)
try:
members = build_index_provider().get_members(index)
except ValueError as exc:
raise _fail(str(exc)) from exc
if not members:
raise _fail(
f"Scraped zero members for index {index!r}; nothing to fetch. "
"Check network access, or whether Wikipedia's table layout changed."
)
try:
index_repository.upsert(index, members, now=now)
except Exception as exc: # noqa: BLE001 - surface any DB failure readably
raise _fail(f"Could not record membership for index {index!r}: {exc}") from exc
typer.echo(f"{len(members)} members recorded for index {index!r}.")
max_age = timedelta(hours=discovery.analyst_scan_max_age_hours)
fresh = consensus_repository.latest_for_symbols(members, max_age=max_age, now=now)
to_fetch = [m for m in members if m not in fresh]
analysts = build_analyst_provider()
failures: list[str] = []
fetched = 0
for position, ticker in enumerate(to_fetch):
try:
opinion = analysts.get_opinion(ticker)
if opinion is not None:
consensus_repository.append(
ticker, opinion, fetched_at=now, source="yfinance"
)
except Exception as exc: # noqa: BLE001 - one symbol must not stop the scan
failures.append(ticker)
typer.echo(f"{ticker:<8}FAILED {exc}")
continue
if opinion is None:
typer.echo(f"{ticker:<8}no analyst coverage")
else:
typer.echo(f"{ticker:<8}recorded (mean={opinion.recommendation_mean})")
fetched += 1
if position < len(to_fetch) - 1:
time.sleep(_SCAN_UNIVERSE_DELAY_SECONDS)
typer.echo(
f"{fetched} fetched, {len(members) - len(to_fetch)} already fresh, "
f"{len(failures)} failed, out of {len(members)} members in index {index!r}."
)
if failures:
raise _fail(
f"{len(failures)} symbol(s) left without a fetch: {', '.join(failures)}. "
"Re-run to fill the gap."
)
[docs]
@app.command("analyst-health")
def analyst_health(
symbols: str = typer.Option(
"",
"--symbols",
help="Comma-separated symbols to score. Defaults to fixed_tickers.",
),
rank_metric: str = typer.Option(
METRIC_STRONG_BUY,
"--rank-metric",
help=f"Rank by '{METRIC_STRONG_BUY}' or '{METRIC_RECOMMENDATION_MEAN}'.",
),
rank_limit: int = typer.Option(
20, "--rank-limit", help="How many rank-persistence rows to show."
),
) -> None:
"""A composite analyst health score, plus a rank-persistence screen
(issue #39) — both computed over data this app already fetched and
stored, and neither wired into any order-affecting path.
The health score (`calculate_health_score`, reimplemented in `Decimal`
from a formula found in `../stock-screener`) is scored per symbol from
the newest persisted `analyst_consensus_history` opinion
(`AnalystConsensusRepository.latest_for_symbols`, the same read
`discover_symbols()`'s analyst-scan source uses) and the most recently
cached bar close (`BarCache`, the same source discovery's own
price/volume filters read from) — never a fresh live analyst or price
call this command triggers itself.
The rank-persistence screen ranks every symbol in `analyst_consensus_
history` against its peers at each historical `trader scan-universe`
fetch, then reports the mean and standard deviation of each symbol's
rank across every fetch it appeared in — closing the gap `docs/
ideas.md` flagged: the consensus delta is stored but was never surfaced.
"""
if rank_metric not in (METRIC_STRONG_BUY, METRIC_RECOMMENDATION_MEAN):
raise _fail(
f"Unknown --rank-metric {rank_metric!r}; expected "
f"'{METRIC_STRONG_BUY}' or '{METRIC_RECOMMENDATION_MEAN}'."
)
config = _load_config()
consensus_repository = build_analyst_consensus_repository(config)
discovery = build_discovery_settings()
cache = build_bar_cache(config)
watchlist_symbols = [s.strip().upper() for s in symbols.split(",") if s.strip()]
if not watchlist_symbols:
watchlist_symbols = [s.upper() for s in load_watchlist()]
try:
now = datetime.now(UTC)
opinions = consensus_repository.latest_for_symbols(
watchlist_symbols,
max_age=timedelta(hours=discovery.analyst_scan_max_age_hours),
now=now,
)
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not read analyst consensus ({_database_reason(exc)}). "
"If the database is missing or predates this version, run "
"'trader migrate'."
) from exc
headers = [
"SYMBOL",
"SCORE",
"TIER",
"UPSIDE %",
"ANALYSTS",
"BUY CANDIDATE",
"REVIEW FLAG",
]
rows: list[list[object]] = []
for symbol in watchlist_symbols:
try:
bars = _recent_bars(cache, symbol, discovery)
except Exception: # noqa: BLE001 - one symbol's price outage must not lose the rest
_logger.warning("analyst-health: could not price %s", symbol, exc_info=True)
bars = []
if not bars:
rows.append(
[symbol, MISSING, "no cached price", MISSING, MISSING, MISSING, MISSING]
)
continue
result = calculate_health_score(opinions.get(symbol), bars[-1].close)
if result is None:
rows.append(
[
symbol,
MISSING,
"no fresh consensus",
MISSING,
MISSING,
MISSING,
MISSING,
]
)
continue
rows.append(
[
symbol,
result.score,
result.tier,
_fmt_pct(result.upside_pct) if result.upside_pct is not None else MISSING,
result.total_analysts,
"yes" if result.is_buy_candidate else "",
"yes" if result.is_review_flag else "",
]
)
typer.echo("Health score:")
typer.echo(build_table(headers, rows))
try:
history_rows = consensus_repository.history()
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not read analyst consensus history ({_database_reason(exc)}). "
"If the database is missing or predates this version, run "
"'trader migrate'."
) from exc
points = [
ConsensusHistoryPoint(
symbol=row.symbol,
fetched_at=row.fetched_at,
strong_buy=row.strong_buy,
recommendation_mean=row.recommendation_mean,
)
for row in history_rows
]
stability = compute_rank_stability(points, metric=rank_metric)[:rank_limit]
typer.echo(f"\nRank-persistence screen (metric={rank_metric}):")
if not stability:
typer.echo("No analyst-scan history yet — run 'trader scan-universe' first.")
return
stability_headers = ["SYMBOL", "MEAN RANK", "RANK STDDEV", "SNAPSHOTS"]
stability_rows: list[list[object]] = [
[
entry.symbol,
entry.mean_rank.quantize(PERCENT_QUANTUM),
entry.stddev_rank.quantize(PERCENT_QUANTUM)
if entry.stddev_rank is not None
else MISSING,
entry.appearances,
]
for entry in stability
]
typer.echo(build_table(stability_headers, stability_rows))
[docs]
@app.command("build-focus-list")
def build_focus_list(
index: str = typer.Option(
ANALYST_SCAN_INDEX, help="Which index to build the focus list from."
),
) -> None:
"""Compute tomorrow's focus list from analyst-scan candidates (issue #44).
Separate from `trader scan-universe` on purpose (Decision 2 of the design
doc): that command refreshes `index_membership`/`analyst_consensus_history`,
the two cheap tables this command only ever *reads* — never a yfinance
call at index scale. Run `scan-universe` first, the same night
(Decision 3); this command degrades to nothing useful against stale or
absent membership, which is why an empty read fails loudly below rather
than silently writing an empty focus list.
Reuses `discover_symbols()`'s existing analyst-scan branch
(`_scan_analyst_candidates` + `_evaluate_and_accept`) rather than
reimplementing candidate selection — called with `themes=[]` so only that
branch runs, `search` never invoked. The same
`listing_gate`/`asset_gate`/`opinion_gate`/price-volume chain every
theme-search candidate already clears applies here, unchanged, including a
fresh live `analysts.get_opinion` call per candidate.
Meant to run off market hours, on its own cadence (cron/launchd, or by
hand) — never inside `trader run`'s daemon loop, the same framing
`scan-universe` already documents for itself. Does not take the issue #43
single-instance lock (Decision 5): no order path, no LLM call, and its
only write is an additive upsert into `focus_list`, a table the daemon
never touches.
Every candidate's own lookup failure is isolated by `discover_symbols()`
itself (unchanged discipline) and costs only that candidate. The
membership and consensus reads are re-checked directly here, ahead of
`discover_symbols()`, so a broken read is a *visible*, non-zero-exit
failure to an operator instead of silently degrading to "zero candidates
found" — same discipline as `scan-universe`'s own "zero members scraped"
check, extended to the consensus read.
"""
config = _load_config()
discovery = build_discovery_settings()
index_repository = build_index_membership_repository(config)
consensus_repository = build_analyst_consensus_repository(config)
focus_list_repository = build_focus_list_repository(config)
now = datetime.now(UTC)
try:
members = index_repository.members(index)
except Exception as exc: # noqa: BLE001 - surface any DB failure readably
raise _fail(
f"Could not read index membership for index {index!r}: {exc}"
) from exc
if not members:
raise _fail(
f"No membership recorded for index {index!r}; nothing to build a "
"focus list from. Run 'trader scan-universe' first."
)
try:
consensus_repository.latest_for_symbols(
members,
max_age=timedelta(hours=discovery.analyst_scan_max_age_hours),
now=now,
)
except Exception as exc: # noqa: BLE001 - surface any DB failure readably
raise _fail(f"Could not read analyst consensus history: {exc}") from exc
try:
fixed = load_watchlist()
except TraderError as exc:
raise _fail(str(exc)) from exc
broker = build_broker(config)
cache = build_bar_cache(config)
candidates = discover_symbols(
settings=discovery.model_copy(update={"themes": []}),
search=build_search_provider(),
analysts=build_analyst_provider(),
broker=broker,
bars_for=lambda s: _recent_bars(cache, s, discovery),
fixed_symbols=fixed,
now=now,
index_repository=index_repository,
consensus_repository=consensus_repository,
)
try:
clock = broker.get_clock()
except Exception as exc: # noqa: BLE001 - surface any broker failure readably
raise _fail(f"Could not read the market clock: {exc}") from exc
# Decision 7: the session the batch is building the list *for*, not
# wall-clock UTC midnight when the batch happened to run.
trading_day = clock.next_open.date()
try:
focus_list_repository.upsert(trading_day, candidates, generated_at=now)
except Exception as exc: # noqa: BLE001 - surface any DB failure readably
raise _fail(f"Could not record the focus list for {trading_day}: {exc}") from exc
typer.echo(
f"{len(candidates)} candidate(s) recorded to the focus list for "
f"{trading_day} (index {index!r}, {len(members)} members scanned)."
)
#: The pre-registered sample (issue #7 `note_3689814772`). Fixed here, never
#: read from `config/watchlist.yaml`: `fixed_tickers` can change, and this
#: sample must not — a config edit must never silently change what a run
#: measured. Dense: AAPL, GOOG, SPY, PLTR, NET, TOST. Thin: ANNX, ITRG, UUUU.
_REPLAY_SYMBOLS = ("AAPL", "ANNX", "GOOG", "ITRG", "NET", "PLTR", "SPY", "TOST", "UUUU")
_REPLAY_TRAIN_SPAN = (date(2023, 8, 7), date(2025, 8, 6))
_REPLAY_HOLDOUT_START = date(2025, 8, 7)
_REPLAY_QUOTAS = (40, 20)
#: The end cut is 21 trading days before the last cached bar, off the
#: session grid — never a fixed calendar date, which would silently drift
#: relative to whatever history happens to be cached.
_REPLAY_END_CUT_TRAILING_SESSIONS = 21
[docs]
@app.command("replay-score")
def replay_score(
out: Path = typer.Option(Path("data/replays"), help="Where to write artifacts."),
resume: bool = typer.Option(False, "--resume", help="Skip rows already in the CSV."),
limit: int | None = typer.Option(
None, help="Smoke-test the first N rows of the pre-registered order."
),
dry_run: bool = typer.Option(
False,
"--dry-run",
help="Build every prompt and exercise both point-in-time guards; call no model.",
),
) -> None:
"""Replay the pre-registered decision sample and score it (issue #7 step 4).
Reads the news archive and the bar cache; calls Ollama (unless
`--dry-run`); writes a per-row CSV and a JSON summary under `--out`.
Places no orders and never contacts the broker — the replay drives
`LlmStrategy` directly with an archive-backed news provider and a fixed
clock, never `OrderExecutor`. Expect ~1.7-1.9 hours for the full 540 rows
at the configured `num_ctx`; `--dry-run` costs nothing but a database
read and exercises the point-in-time guards over the whole sample first.
Everything about the sample — the 9 symbols, the train/holdout split, the
40+20 per-symbol quota, the 72h eligibility window — is pre-registered
and fixed in code, not read from `config/watchlist.yaml`: a config edit
must never silently change what a run measures.
"""
config = _load_config()
bar_repository = build_bar_repository(config)
params = llm_strategy_params()
window_hours = float(params.get("max_news_age_hours") or DEFAULT_NEWS_WINDOW_HOURS)
# The trading-day grid every horizon and every eligibility check is
# counted against, off SPY's own cached bars — never a weekday rule,
# which is wrong at every market holiday.
all_spy_bars = bar_repository.load_bars(
"SPY", "1d", datetime(2023, 1, 1, tzinfo=UTC), datetime.now(UTC)
)
if not all_spy_bars:
raise _fail(
"No cached SPY '1d' bars found. Fetch bar history for the nine "
"replay symbols back to 2023-08-07 (e.g. `trader fetch-bars`) "
"before running the replay."
)
grid = session_grid(all_spy_bars)
if len(grid) <= _REPLAY_END_CUT_TRAILING_SESSIONS:
raise _fail(
f"Only {len(grid)} cached SPY sessions; need more than "
f"{_REPLAY_END_CUT_TRAILING_SESSIONS} to cut the replay window."
)
end_cut = grid[-1 - _REPLAY_END_CUT_TRAILING_SESSIONS]
holdout_span = (_REPLAY_HOLDOUT_START, end_cut)
if end_cut < _REPLAY_HOLDOUT_START:
raise _fail(
f"The bar-cut end {end_cut} is before the holdout start "
f"{_REPLAY_HOLDOUT_START}; not enough cached history yet."
)
archive_repository = build_news_archive_repository(config)
rows, shortfalls = build_sample(
archive_repository,
_REPLAY_SYMBOLS,
grid,
train=_REPLAY_TRAIN_SPAN,
holdout=holdout_span,
quotas=_REPLAY_QUOTAS,
window_hours=window_hours,
)
if limit is not None:
rows = rows[:limit]
n_train = sum(1 for r in rows if r.split == "train")
n_holdout = sum(1 for r in rows if r.split == "holdout")
typer.echo(f"Sample: {len(rows)} rows ({n_train} train / {n_holdout} holdout)")
for shortfall in shortfalls:
typer.echo(
f" shortfall: {shortfall.symbol:<6}{shortfall.split:<8}"
f"wanted {shortfall.wanted} got {shortfall.got} "
f"(missing {shortfall.missing})"
)
if not rows:
raise _fail(
"No eligible decision dates in the pre-registered sample. Has "
"`trader backfill-news` run for these symbols?"
)
runner = build_replay_runner(config)
report = runner.run(rows, out, resume=resume, dry_run=dry_run, shortfalls=shortfalls)
typer.echo(f"Wrote {report.n_scored} scored rows to {report.csv_path}")
typer.echo(f"Summary: {report.json_path}")
_print_replay_summary(Path(report.json_path))
[docs]
@app.command("replay-rescore")
def replay_rescore(
csv_path: Path = typer.Argument(
..., help="An already-written replay_rows.csv from a prior replay-score run."
),
out: Path | None = typer.Option(
None,
help="Where to write the rescored replay_summary.json; defaults to "
"csv_path's own directory.",
),
) -> None:
"""Re-score an existing replay CSV under the current formula (issue #29).
Calls no model and fetches nothing — every replayed decision (action,
confidence, and every horizon's forward return) is already in `csv_path`.
Use this instead of a fresh `replay-score` run whenever the *scoring*
code changes and an already-collected sample needs to be re-judged
(edge_A's degeneracy fix is the first case; it cost ~1.7-1.9 hours of
Ollama calls to collect and must not cost that again to re-read).
Reads `replay_summary.json` next to `csv_path`, if present, only for
provenance (`seed`, `fetch_dates`, `dry_run`, `shortfalls`) — never for
scored data, which comes from the CSV alone. `edge_grid_market_adjusted`
is **not recomputed**: SPY's forward-price grid needed for it cannot be
reconstructed from the CSV and this command makes no market-data fetch,
so that section reports null. Everything else is fully reprocessed.
"""
out_dir = out or csv_path.parent
old_summary_path = csv_path.parent / "replay_summary.json"
seed = 7
fetch_dates: dict[str, str] = {}
dry_run = False
shortfalls: list[SampleShortfall] = []
if old_summary_path.exists() and old_summary_path != (
out_dir / "replay_summary.json"
):
old = json.loads(old_summary_path.read_text())
seed = int(old.get("seed", seed))
fetch_dates = dict(old.get("fetch_dates") or {})
dry_run = bool(old.get("dry_run", dry_run))
shortfalls = [
SampleShortfall(
symbol=s["symbol"], split=s["split"], wanted=s["wanted"], got=s["got"]
)
for s in old.get("shortfalls") or []
]
report = rescore_from_csv(
csv_path,
out_dir,
seed=seed,
fetch_dates=fetch_dates,
shortfalls=shortfalls,
dry_run=dry_run,
)
typer.echo(f"Rescored {report.n_scored} rows from {csv_path}")
typer.echo(f"Summary: {report.json_path}")
_print_replay_summary(Path(report.json_path))
def _print_replay_summary(json_path: Path) -> None:
"""Print the headline result, its CI, and any degeneracy warning (issue #29).
Prints to stdout, never only to the log — the same reasoning `CycleReport`
already carries for `unprotected`/`double_protected`: a warning nobody
watching the command sees is not a warning that reached anyone.
"""
if not json_path.exists():
# A test double standing in for `ReplayRunner`/`rescore_from_csv`
# reports a `json_path` it never actually wrote — there is nothing
# to summarize, and that is not this command's failure to report.
return
summary = json.loads(json_path.read_text())
metric = summary.get("headline_metric", "edge_a")
headline = summary.get("headline_edge_n5_holdout") or summary.get(
"headline_edge_a_n5_holdout", {}
)
interval = summary.get("headline_interval_n5_holdout", {})
typer.echo(
f"Headline ({metric}, N=5, holdout): edge={headline.get('edge')} "
f"(model={headline.get('model_return')}, null={headline.get('null_return')}, "
f"n_acting={headline.get('n_acting')}, n_rows={headline.get('n_rows')})"
)
low, high = interval.get("low"), interval.get("high")
if low is not None and high is not None:
includes_zero = low <= 0.0 <= high
typer.echo(
f"95% CI: [{low}, {high}] over {interval.get('resamples_used')} of "
f"{interval.get('draws')} draws — {'includes' if includes_zero else 'excludes'} zero"
)
else:
typer.echo(
"95% CI: no resamples produced a value (empty row set at this horizon)."
)
diagnostic = summary.get("edge_a_entry_only_diagnostic_n5_holdout")
if diagnostic and diagnostic.get("degenerate"):
typer.echo(
"WARNING: edge_A is structurally degenerate on this sample (zero SELL "
"actions among the acting rows at N=5 holdout) — its edge "
f"({diagnostic.get('edge')}) is 0.0 by construction, not by measurement, "
"and is reported only as a diagnostic, never as the headline test. "
"See issue #29."
)
null_sentence = summary.get("null_sentence")
if null_sentence:
typer.echo(null_sentence)
[docs]
@app.command()
def strategies() -> None:
"""List the strategies configured in config/strategies.yaml."""
try:
configured = load_strategies()
except TraderError as exc:
raise _fail(str(exc)) from exc
if not configured:
typer.echo("No strategies configured.")
return
headers = ["ID", "TYPE", "PARAMS"]
rows = [[entry.id, entry.type, entry.params] for entry in configured]
typer.echo(build_table(headers, rows))
[docs]
@app.command()
def backtest(
strategy: str = typer.Option(..., help="Strategy id from strategies.yaml."),
ticker: str = typer.Option(..., help="Symbol to test."),
start: datetime = typer.Option(..., formats=["%Y-%m-%d"]),
end: datetime = typer.Option(..., formats=["%Y-%m-%d"]),
) -> None:
"""Run one strategy against one symbol over a date range."""
config = _load_config()
try:
configured = {c.id: c for c in load_strategies()}
except TraderError as exc:
raise _fail(str(exc)) from exc
entry = configured.get(strategy)
if entry is None:
raise _fail(
f"Unknown strategy id {strategy!r}. Configured: {', '.join(sorted(configured)) or 'none'}."
)
symbol = ticker.upper()
window_start = start.replace(tzinfo=UTC)
window_end = end.replace(tzinfo=UTC)
# Build before fetching: an invalid parameter is a config error, and there
# is no reason to spend a network round trip discovering it.
try:
built = build_strategy(entry)
except TraderError as exc:
raise _fail(str(exc)) from exc
# Each broad `except` below is scoped to the one call whose failure it
# describes. A single handler around the whole body reported a bad strategy
# parameter as a database problem and sent the user to 'trader setup init-db'.
try:
bars = build_bar_cache(config).ensure(symbol, window_start, window_end)
except TraderError as exc:
raise _fail(str(exc)) from exc
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not read or write cached bars ({type(exc).__name__}: {exc}). "
"If the database is missing or predates this version, run "
"'trader setup init-db'."
) from exc
try:
result = run_backtest(built, symbol, bars, BacktestConfig())
except TraderError as exc:
raise _fail(str(exc)) from exc
try:
run_id = build_backtest_repository(config).save(
result, window_start, window_end, entry.params
)
except TraderError as exc:
raise _fail(str(exc)) from exc
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not save the backtest run ({type(exc).__name__}: {exc}). "
"If the database is missing or predates this version, run "
"'trader setup init-db'."
) from exc
typer.echo(f"{result.strategy_id} on {result.symbol} ({len(bars)} bars)")
typer.echo(f" starting value: {result.starting_value}")
typer.echo(f" ending value: {result.ending_value}")
typer.echo(f" total return: {result.total_return_pct:.2f}%")
typer.echo(f" max drawdown: {result.max_drawdown_pct:.2f}%")
typer.echo(f" wins / losses: {result.win_count} / {result.loss_count}")
typer.echo(f" trades: {len(result.trades)}")
# The baseline is printed with the result, never behind a flag: a return
# shown alone gets quoted alone, and then a strategy that gave up 54 points
# to the symbol it traded reads as a 21% success.
typer.echo(f"\n buy & hold: {result.benchmark_return_pct:.2f}%")
typer.echo(f" buy & hold value:{result.benchmark_ending_value:>12.2f}")
typer.echo(f" excess vs hold: {result.excess_return_pct:+.2f}%")
typer.echo(f"\nSaved as backtest run #{run_id}.")
# Stretch goal (issue #19 / requirements §12.1): a market-index comparison
# alongside the backtest's own same-symbol buy-and-hold above. Best-effort
# and never fatal — a benchmark ticker's data outage must not turn an
# otherwise-successful, already-saved backtest into a CLI failure, the same
# isolation discipline discovery uses per theme.
reporting_config = build_reporting_config()
other_tickers = [t for t in reporting_config.benchmark_tickers if t != symbol]
if other_tickers:
typer.echo("\nMarket index comparison (buy & hold over the same dates):")
for index_ticker in other_tickers:
try:
index_bars = build_bar_cache(config).ensure(
index_ticker, window_start, window_end
)
index_result = benchmark_return(
index_ticker, index_bars, result.starting_value
)
except Exception as exc: # noqa: BLE001 - best-effort, see comment above
typer.echo(f" {index_ticker:<6}FAILED {type(exc).__name__}: {exc}")
continue
excess = result.total_return_pct - index_result.return_pct
typer.echo(
f" {index_ticker:<6}{index_result.return_pct:>+8.2f}% "
f"(strategy {excess:+.2f}pp "
f"{'ahead of' if excess >= 0 else 'behind'} {index_ticker})"
)
[docs]
@app.command()
def positions(
risk: bool = typer.Option(
False,
"--risk",
help=(
"Also print sector concentration and return-correlation across "
"held positions, plus discovered-candidate sector distribution "
"(issue #65). Measurement only, never a gate. Costs extra "
"yfinance calls (sector lookups, and bar fetches for any symbol "
"not already cached), so it is opt-in rather than printed by "
"default."
),
),
) -> None:
"""Show open positions."""
config = _load_config()
try:
open_positions = build_broker(config).get_positions()
except TraderError as exc:
raise _fail(str(exc)) from exc
if not open_positions:
typer.echo("No open positions.")
else:
# Best-effort: this command has never required a database before,
# and must not start failing over a "days held" column. See
# `_open_round_trips`.
open_trips = _open_round_trips(build_outcome_repository(config), open_positions)
typer.echo(_positions_table(open_positions, open_trips, datetime.now(UTC)))
if risk:
_print_risk_report(config, open_positions, datetime.now(UTC))
def _run_once_body(
config: EffectiveTradingConfig, dry_run: bool, strategy: str | None
) -> None:
"""The real run-once implementation.
Runs inside the single-instance lock `run_once_command` (below) acquires
before calling this — split out so the lock-acquisition wrapper stays
thin and the CLI-wiring test can prove the lock gates this body without
ever reaching a real cycle.
"""
_record_deployed_version(config)
try:
pipeline_config = build_pipeline_config()
symbols = load_watchlist()
configured = {c.id: c for c in load_strategies()}
except TraderError as exc:
raise _fail(str(exc)) from exc
if not configured:
raise _fail("No strategies configured in config/strategies.yaml.")
if not symbols:
raise _fail("No tickers in the watchlist at config/watchlist.yaml.")
entries = list(configured.values())
if strategy is not None:
chosen = configured.get(strategy)
if chosen is None:
raise _fail(
f"Unknown strategy id {strategy!r}. "
f"Configured: {', '.join(sorted(configured))}."
)
# Narrows which strategies run. It does NOT promote a shadow strategy
# to live: a CLI flag that grants trading authority is the foot-gun
# this project avoids elsewhere. Going live is a config edit.
entries = [chosen]
if not any(e.mode is StrategyMode.TRADING for e in entries):
typer.echo(
"All strategies are in shadow mode — decisions will be recorded "
"but no orders placed. Set `mode: trading` on one strategy in "
"config/strategies.yaml to trade.\n"
)
if dry_run:
typer.echo("DRY RUN — no orders will be submitted.\n")
try:
report = execute_cycle(
config=config,
pipeline_config=pipeline_config,
strategy_entries=entries,
symbols=symbols,
dry_run=dry_run,
)
except TraderError as exc:
raise _fail(str(exc)) from exc
except DatabaseError as exc:
# `round_trips` is new in this slice, and `create_db_engine` neither
# creates nor migrates anything. An existing database that predates it
# reaches `open_round_trip_for` on the first held position — no fills
# needed — and used to dump a raw `OperationalError` and a
# sqlalche.me link. `migrate`, not `init-db`: this is an existing
# database gaining a new table. Caught narrowly on `DatabaseError`
# rather than `Exception` so a signature drift between `execute_cycle`
# and `run_once` still surfaces as the `TypeError` it is instead of
# being misreported as a schema problem.
raise _fail(
f"The cycle could not read or write the database "
f"({_database_reason(exc)}). "
"If the database predates this version, run 'trader migrate'."
) from exc
if report.halted:
typer.echo("Daily loss limit reached — no new entries this cycle.\n")
for outcome in report.outcomes:
# "-" rather than "" for an unranked outcome: most rows are unranked
# (only entry candidates are ever ranked), and a blank column would
# make a working ranker and a silently failing one print identically.
rank = "-" if outcome.rank is None else str(outcome.rank)
typer.echo(
f"{outcome.symbol:<8}{outcome.strategy_id:<20}"
f"{outcome.action:<10}{rank:>4} {outcome.reason}"
)
if report.entries_cancelled:
count = len(report.entries_cancelled)
noun = "entry" if count == 1 else "entries"
typer.echo(
f"\nCancelled {count} stale {noun} near the close: "
f"{', '.join(report.entries_cancelled)}."
)
# Money-tracking bookkeeping rather than a §8 safety field, but printed for
# the same reason: logging here is file-only, and an operator watching
# `run-once` is the only reader who can tell whether issue #1's window is
# actually narrowing.
if report.orders_settled:
count = len(report.orders_settled)
noun = "order" if count == 1 else "orders"
typer.echo(
f"\nSettled {count} terminal unfilled {noun}: "
f"{', '.join(report.orders_settled)}."
)
if report.stops_placed:
count = len(report.stops_placed)
noun = "trailing stop" if count == 1 else "trailing stops"
typer.echo(f"\nPlaced {count} {noun}: {', '.join(report.stops_placed)}.")
if report.stops_ratcheted:
count = len(report.stops_ratcheted)
noun = "stop" if count == 1 else "stops"
typer.echo(f"Raised {count} {noun}: {', '.join(report.stops_ratcheted)}.")
# The two safety fields go to *stdout*, not only to the log. Requirements
# §8: "the app must never terminate quietly while it believes money is
# unprotected." Logging in this app is file-only — there is no
# `StreamHandler` anywhere in `src/` — so every §8 WARNING `run_once`
# emits misses the terminal entirely, and stdout is the path an operator
# actually watches. Printed even though the command still exits 0:
# changing the exit code is a separate decision, but staying silent is not
# a decision this section is allowed to make.
if report.unprotected:
typer.echo(
f"\nWARNING: {len(report.unprotected)} position(s) are UNPROTECTED "
f"(no open protective order at the broker): "
f"{', '.join(report.unprotected)}."
)
if report.double_protected:
typer.echo(
f"\nWARNING: {len(report.double_protected)} position(s) carry more "
f"than one open protective SELL: "
f"{', '.join(report.double_protected)}. If two execute, the second "
f"sells shares this account does not hold."
)
if report.outstanding_entries:
typer.echo(
f"\nWARNING: {len(report.outstanding_entries)} entry order(s) are "
f"still unfilled at the broker: "
f"{', '.join(report.outstanding_entries)}. Nothing is watching "
f"them once this command exits."
)
if report.snapshot_id is not None:
typer.echo(f"Saved snapshot #{report.snapshot_id}.")
[docs]
@app.command("run-once")
def run_once_command(
dry_run: bool = typer.Option(
False, "--dry-run", help="Log intended orders without submitting them."
),
strategy: str | None = typer.Option(
None,
"--strategy",
help="Strategy id. Defaults to every configured strategy.",
),
) -> None:
"""Walk the watchlist once: evaluate, trade, protect, and snapshot.
Guarded by a single-instance lock scoped to the resolved database path
(issue #43): a second `run`/`run-once` pointed at the same database
fails fast instead of racing this one.
"""
config = _load_config()
try:
with single_instance_lock(config.database_url):
_run_once_body(config, dry_run, strategy)
except InstanceAlreadyRunningError as exc:
raise _fail(str(exc)) from exc
def _run_body(
config: EffectiveTradingConfig, dry_run: bool, max_cycles: int | None
) -> None:
"""The real daemon implementation.
Runs inside the single-instance lock `run` (below) acquires before
calling this — split out so the lock-acquisition wrapper stays thin and
the CLI-wiring test can prove the lock gates this body without ever
starting the daemon loop for real.
"""
_record_deployed_version(config)
logger = logging.getLogger("trader.daemon")
try:
pipeline_config = build_pipeline_config()
daemon_config = build_daemon_config()
symbols = load_watchlist()
entries = load_strategies()
except TraderError as exc:
raise _fail(str(exc)) from exc
if not entries:
raise _fail("No strategies configured in config/strategies.yaml.")
if not symbols:
raise _fail("No tickers in the watchlist at config/watchlist.yaml.")
# Every startup check the daemon needs — schema revision, and (when an
# `llm` strategy is configured) that the model server actually answers —
# lives in `preflight_failures` so it is testable without a runner and so
# a single startup reports *every* problem rather than one restart per
# problem.
def llm_health() -> str:
# `preflight_failures` catches only `TraderError` from this callable,
# so it must not raise anything else. `OllamaProvider.health()`
# already wraps every failure into `LlmError` (a `TraderError`),
# which is what makes that narrow catch safe.
params = llm_strategy_params()
return build_llm_provider(
model=str(params.get("model") or "qwen2.5:32b"),
timeout_seconds=int(params.get("timeout_seconds") or 120),
).health()
failures = preflight_failures(
database_url=config.database_url,
strategy_entries=entries,
llm_health=llm_health,
)
if failures:
for failure in failures:
typer.echo(f"Error: {failure}")
raise typer.Exit(code=1)
if not confirm_live_trading(
is_live=config.is_live,
isatty=sys.stdin.isatty(),
ask=typer.confirm,
):
typer.echo("Aborted.")
raise typer.Exit(code=1)
live = [e.id for e in entries if e.mode is StrategyMode.TRADING]
# Logged before the first cycle, and to stdout as well: an unattended
# process should say what it is empowered to do before it starts doing it.
for key, value in safe_config_summary(config).items():
logger.info("config %s: %s", key, value)
logger.info("cycle interval: %d minute(s)", daemon_config.cycle_interval_minutes)
logger.info("watchlist: %s", ", ".join(symbols))
logger.info("live strategy: %s", live[0] if live else "none (all shadow)")
typer.echo(
f"Starting daemon: {len(symbols)} symbol(s), {len(entries)} strategy(ies), "
f"every {daemon_config.cycle_interval_minutes} minute(s)."
)
typer.echo(
f"Live strategy: {live[0]}."
if live
else "All strategies are in shadow mode — decisions recorded, no orders placed."
)
if dry_run:
typer.echo("DRY RUN — no orders will be submitted.")
typer.echo("Ctrl-C to stop; the current cycle finishes first.\n")
try:
cycle, read_only_broker = build_daemon_cycle(
config=config,
pipeline_config=pipeline_config,
daemon_config=daemon_config,
strategy_entries=entries,
symbols=symbols,
dry_run=dry_run,
)
except TraderError as exc:
raise _fail(str(exc)) from exc
shutdown = ShutdownFlag()
restore_handlers = install_handlers(shutdown)
try:
code = run_forever(
broker=read_only_broker,
cycle=cycle,
schedule=MarketSchedule(daemon_config),
backoff=Backoff(
base_seconds=daemon_config.backoff_base_seconds,
max_seconds=daemon_config.backoff_max_seconds,
),
sleeper=RealSleeper(),
shutdown=shutdown,
config=daemon_config,
max_cycles=max_cycles,
)
finally:
restore_handlers()
if code != 0:
raise typer.Exit(code=code)
typer.echo("Daemon stopped.")
[docs]
@app.command()
def run(
dry_run: bool = typer.Option(
False, "--dry-run", help="Log intended orders without submitting them."
),
max_cycles: int | None = typer.Option(
None,
"--max-cycles",
help="Stop after this many completed cycles. For verification runs.",
),
) -> None:
"""Run continuously: trade during market hours, idle when closed.
Stops cleanly on Ctrl-C or SIGTERM, finishing the cycle it is in first.
Guarded by a single-instance lock scoped to the resolved database path
(issue #43): a second `run`/`run-once` pointed at the same database
fails fast instead of racing this one.
"""
config = _load_config()
try:
with single_instance_lock(config.database_url):
_run_body(config, dry_run, max_cycles)
except InstanceAlreadyRunningError as exc:
raise _fail(str(exc)) from exc
def _no_trips_message(window_start: datetime | None, window_end: datetime | None) -> str:
"""The empty-result hint for `outcomes` when a window excludes every
closed round trip (issue #57 follow-up, finding 3).
`--since-creation` only clears `window_start` — see
`resolve_report_start`'s own precedence — it can never widen
`window_end`, which comes only from an explicit `--end`. Suggesting
`--since-creation` when the window is empty solely because of `--end`
would be a documented no-op, so that suggestion is reserved for the case
where `window_start` is actually still floored; an `--end`-only empty
window instead suggests adjusting `--end`.
"""
range_text = (
f"({window_start.date() if window_start else 'earliest'} -> "
f"{window_end.date() if window_end else 'now'})"
)
if window_start is not None:
return (
f"No closed round trips in this window {range_text}. "
"Pass --since-creation for full history."
)
return f"No closed round trips in this window {range_text}. Try a later --end."
def _outcome_rows(trips: list[RoundTrip]) -> list[list[object]]:
"""One `outcomes` table row per closed round trip (issue #57 follow-up,
finding 2's complexity split)."""
rows: list[list[object]] = []
for trip in trips:
pct = return_pct(trip.realized_pl, cost_basis(trip.entry_price, trip.quantity))
held = days_held(trip.opened_at, trip.closed_at)
avg = avg_daily_return_pct(pct, held) if pct is not None else None
rows.append(
[
trip.symbol,
trip.strategy_id or MISSING,
trip.quantity,
trip.entry_price,
trip.exit_price,
trip.realized_pl,
_fmt_pct(pct),
_fmt_days(held),
_fmt_pct(avg),
trip.closed_at.strftime("%Y-%m-%d"),
]
)
return rows
def _outcome_totals_line(summary: OutcomeSummary) -> str:
"""The 'Total realized P/L / Total return / Win rate' line below
`outcomes`' table. Spans every closed round trip in the window, not just
the `--limit` rows the table shows — see `OutcomeSummary` (issue #57
follow-up, finding 2's complexity split)."""
total_pct = return_pct(summary.total_realized_pl, summary.total_cost_basis)
win_rate = (
(Decimal(summary.win_count) / summary.trip_count * 100).quantize(PERCENT_QUANTUM)
if summary.trip_count
else None
)
win_rate_text = f"{win_rate}%" if win_rate is not None else MISSING
return (
f"\nTotal realized P/L: {summary.total_realized_pl}"
f" Total return: {_fmt_pct(total_pct)}"
f" Win rate: {summary.win_count}/{summary.trip_count} ({win_rate_text})"
)
[docs]
@app.command()
def outcomes(
limit: int = typer.Option(20, help="How many round trips to show."),
start: datetime | None = typer.Option(
None,
formats=["%Y-%m-%d"],
help=(
"Earliest close date to include. Defaults to config/reporting.yaml's "
"default_report_start_date (2026-08-17, unless reconfigured), or all "
"closed round trips if that field is unset. Always wins over "
"--since-creation when both are given."
),
),
end: datetime | None = typer.Option(
None, formats=["%Y-%m-%d"], help="Latest close date to include. Defaults to now."
),
since_creation: bool = typer.Option(
False,
"--since-creation",
help=(
"Ignore config/reporting.yaml's default_report_start_date and show "
"every closed round trip regardless of close date. Overridden by an "
"explicit --start."
),
),
) -> None:
"""Show closed round trips and their realized P/L.
**Default start date (issue #57).** Same rule as `trader performance`:
without `--start`, this windows to `config/reporting.yaml`'s
`default_report_start_date` onward (a dated, deployment-specific finding —
see `docs/operational-timeline.md` — never a code constant), which
excludes round trips CLOSED before the live service was running
cleanly. `--since-creation` opts back into every closed round trip ever
recorded; `--start` (or `--start` together with `--since-creation`)
always wins. `--limit` still caps the page size within whatever window is
in effect — the totals line below the table always spans every closed
trip IN THE WINDOW, not just the page shown, exactly as it did before
this window existed.
"""
config = _load_config()
reporting_config = build_reporting_config()
window_start = resolve_report_start(
explicit_start=start.replace(tzinfo=UTC) if start is not None else None,
since_creation=since_creation,
default_start_date=reporting_config.default_report_start_date,
)
window_end = end.replace(tzinfo=UTC) if end is not None else None
repository = build_outcome_repository(config)
try:
trips = repository.recent(limit=limit, start=window_start, end=window_end)
summary = repository.outcome_summary(start=window_start, end=window_end)
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
# `round_trips` is new in this slice, so a database from before it
# exists is the ordinary upgrade case, not an exotic one. Every other
# persistence-backed command already says this; without it, `outcomes`
# dumped the full SELECT and a sqlalche.me link.
raise _fail(
f"Could not read round trips ({_database_reason(exc)}). "
"If the database is missing or predates this version, run "
"'trader migrate'."
) from exc
if not trips:
if window_start is not None or window_end is not None:
typer.echo(_no_trips_message(window_start, window_end))
else:
typer.echo("No closed round trips yet.")
return
headers = [
"SYMBOL",
"STRATEGY",
"QTY",
"ENTRY",
"EXIT",
"P/L",
"RETURN %",
"DAYS HELD",
"AVG %/DAY (SIMPLE)",
"CLOSED",
]
typer.echo(build_table(headers, _outcome_rows(trips)))
# Deliberately not grouped by strategy: only the live strategy trades, so a
# by-strategy breakdown would imply a comparison this data cannot support.
# See the spec's "What this slice does not deliver".
#
# `summary` spans every closed round trip, not just the `limit` rows
# printed above — a totals line computed only from the visible page would
# silently change value every time `--limit` did. See `OutcomeSummary`.
typer.echo(_outcome_totals_line(summary))
@dataclass(frozen=True, slots=True)
class _SimPortfolioMetrics:
"""One portfolio's row of `sim-report` numbers, computed once and reused
for both its own row and (when it is a normal portfolio's AbB sibling)
the delta column (issue #42).
`equity`/`return_pct`/`return_excl_stops_pct` are `None` (issue #57) only
in the WINDOWED case where the window contains fewer than 2 equity
snapshots — the same "cannot show a return from one point" reasoning
`trader.performance.benchmark.account_return` uses, extended here so a
narrow `--start`/`--end` degrades to MISSING cells rather than a
fabricated 0.00%.
"""
equity: Decimal | None
return_pct: Decimal | None
return_excl_stops_pct: Decimal | None
max_drawdown_pct: Decimal | None
wins: int
losses: int
stop_exits: int
#: A return needs two points; see `_sim_portfolio_metrics`.
_MIN_POINTS_FOR_A_WINDOWED_RETURN = 2
def _in_window(ts: datetime, start: datetime | None, end: datetime | None) -> bool:
"""Both bounds inclusive — same convention as `SnapshotRepository
.snapshots_in_range` and `OutcomeRepository.recent`'s new `start`/`end`.
`start=end=None` (the unwindowed case) is always `True`, which is what
lets `_sim_portfolio_metrics` filter unconditionally below without a
separate branch for "no window given"."""
if start is not None and ts < start:
return False
return not (end is not None and ts > end)
def _live_metrics(
repository: SimulationRepository,
portfolio: SimPortfolio,
non_stop_pl: Decimal,
) -> tuple[Decimal, Decimal | None, Decimal | None]:
"""The unwindowed case: `equity` is the portfolio's current live
mark-to-market (cash + open positions at `last_price`), and `return_pct`
is computed against `portfolio.starting_cash` — the moment-in-time "how is
this strategy doing right now" view `sim-report` shipped with, unchanged
by issue #57's windowing.
`equity` is always a concrete `Decimal` here (never MISSING) — computed as
a plain non-Optional local so `total_return_pct` gets a non-Optional
argument (issue #57 follow-up, finding 1's mypy fix). The caller widens it
into `_SimPortfolioMetrics`'s `Decimal | None` field, which the windowed
case genuinely needs to be `None` for.
"""
positions = repository.all_positions(portfolio.id)
equity: Decimal = portfolio.cash + sum(
(p.quantity * p.last_price for p in positions), Decimal(0)
)
return_pct = total_return_pct(portfolio.starting_cash, equity)
return_excl_stops_pct = total_return_pct(
portfolio.starting_cash, portfolio.starting_cash + non_stop_pl
)
return equity, return_pct, return_excl_stops_pct
def _windowed_metrics(
curve_values: list[Decimal],
non_stop_pl: Decimal,
) -> tuple[Decimal | None, Decimal | None, Decimal | None]:
"""The windowed case: recompute `equity`/`return_pct` from the WINDOW's
own first/last `sim_equity_snapshots` row — mirroring exactly how `trader
performance` computes `account_return` from the first/last
`account_snapshots` row in its window. This does NOT re-run the
simulator; see `_sim_portfolio_metrics`'s docstring for the full design
decision.
Fewer than `_MIN_POINTS_FOR_A_WINDOWED_RETURN` equity snapshots fell
inside the window: there is no pair of points to compute a return from.
Both returns are `None` (MISSING), never a fabricated 0.00% — the same
"unmeasurable, not flat" reasoning `account_return` uses. The lone
snapshot (if any) still displays as the equity value, since that much
genuinely is known.
"""
if len(curve_values) >= _MIN_POINTS_FOR_A_WINDOWED_RETURN:
window_start_equity, window_end_equity = curve_values[0], curve_values[-1]
return_pct = total_return_pct(window_start_equity, window_end_equity)
return_excl_stops_pct = total_return_pct(
window_start_equity, window_start_equity + non_stop_pl
)
return window_end_equity, return_pct, return_excl_stops_pct
equity = curve_values[0] if curve_values else None
return equity, None, None
def _trip_aggregates(trips: list[SimRoundTrip]) -> tuple[Decimal, int, int, int]:
"""`non_stop_pl`/`wins`/`losses`/`stop_exits` — the four numbers both
`_live_metrics` and `_windowed_metrics` need, computed once from the
already-windowed `trips` list rather than twice (issue #57 follow-up,
finding 2)."""
non_stop_pl = sum(
(t.realized_pl for t in trips if t.exit_reason != EXIT_TRAILING_STOP),
Decimal(0),
)
wins, losses = win_loss([t.realized_pl for t in trips])
stop_exits = sum(1 for t in trips if t.exit_reason == EXIT_TRAILING_STOP)
return non_stop_pl, wins, losses, stop_exits
def _sim_portfolio_metrics(
repository: SimulationRepository,
portfolio: SimPortfolio,
start: datetime | None = None,
end: datetime | None = None,
) -> _SimPortfolioMetrics:
"""`start`/`end` window this portfolio's numbers (issue #57's `sim-report`
windowing).
**Design decision (issue #57), not a re-run of the simulator**: windowing
a shadow portfolio means trimming its already-recorded
`sim_equity_snapshots` and `sim_round_trips` to `[start, end]` and
recomputing return from the WINDOW's own first/last equity snapshot —
mirroring exactly how `trader performance` computes `account_return` from
the first/last `account_snapshots` row in its window. It does NOT mean
replaying the simulator from a different starting point, which would be a
materially larger change (the simulator's whole reason to exist is
walking every historical bar from inception, and a "start mid-history"
run has no well-defined starting cash or open positions to seed from).
When `start` and `end` are both `None` (the unwindowed case — no
`--start`/`--end`/`--since-creation` in effect, or an unset
`default_report_start_date`), this returns EXACTLY what it always did —
see `_live_metrics`. Windowing only changes the calculation when a bound
is actually in effect (`_windowed_metrics`), so a plain `trader
sim-report` continues to match every pre-issue-#57 assertion.
The two branches are structurally disjoint (live mark-to-market vs.
windowed first/last-snapshot recompute), so they live in their own
helpers; what they share — `non_stop_pl`/`wins`/`losses`/`stop_exits`
(`_trip_aggregates`) — is computed once here and passed into whichever
branch is taken. This function is now a thin dispatcher: filter, compute
the shared aggregates, delegate, assemble the result.
"""
windowed = start is not None or end is not None
curve = [
snap
for snap in repository.equity_curve(portfolio.id)
if _in_window(snap.bar_ts, start, end)
]
trips = [
t
for t in repository.round_trips(portfolio.id)
if _in_window(t.closed_at, start, end)
]
curve_values = [snap.equity for snap in curve]
non_stop_pl, wins, losses, stop_exits = _trip_aggregates(trips)
equity: Decimal | None
return_pct: Decimal | None
return_excl_stops_pct: Decimal | None
if not windowed:
equity, return_pct, return_excl_stops_pct = _live_metrics(
repository, portfolio, non_stop_pl
)
else:
equity, return_pct, return_excl_stops_pct = _windowed_metrics(
curve_values, non_stop_pl
)
return _SimPortfolioMetrics(
equity=equity,
return_pct=return_pct,
return_excl_stops_pct=return_excl_stops_pct,
max_drawdown_pct=max_drawdown_pct(curve_values) if curve_values else None,
wins=wins,
losses=losses,
stop_exits=stop_exits,
)
def _sim_portfolio_row(
label: str, metrics: _SimPortfolioMetrics, delta: str
) -> list[object]:
return [
label,
metrics.equity if metrics.equity is not None else MISSING,
_fmt_pct(metrics.return_pct),
_fmt_pct(metrics.return_excl_stops_pct),
f"{metrics.max_drawdown_pct:.2f}%"
if metrics.max_drawdown_pct is not None
else MISSING,
f"{metrics.wins}/{metrics.losses}",
metrics.stop_exits,
delta,
]
[docs]
@app.command(name="sim-report")
def sim_report(
start: datetime | None = typer.Option(
None,
formats=["%Y-%m-%d"],
help=(
"Earliest bar/close to include. Defaults to config/reporting.yaml's "
"default_report_start_date (2026-08-17, unless reconfigured), or full "
"simulator history if that field is unset. Always wins over "
"--since-creation when both are given."
),
),
end: datetime | None = typer.Option(
None, formats=["%Y-%m-%d"], help="Latest bar/close to include. Defaults to now."
),
since_creation: bool = typer.Option(
False,
"--since-creation",
help=(
"Ignore config/reporting.yaml's default_report_start_date and show "
"each portfolio's full history since the simulator started stepping "
"it. Overridden by an explicit --start."
),
),
) -> None:
"""Show every strategy's shadow-portfolio performance (issue #33), each
paired with its Always-be-Buying (AbB) variant (issue #42).
One row pair per configured strategy — shadow and trading alike, since
the whole point of the simulator is comparing all of them on equal
footing, something the live account (which only the `trading` strategy
ever touches) cannot show. "Return excl. stops" answers "how did this
strategy's own sell signals do", the same ledger split the spec's
acceptance criteria asked for.
The AbB variant — same recorded decisions, sell signals ignored, only the
trailing stop can exit — is shown directly under its strategy's own row,
with "ABB DELTA %" carrying the headline number: how much better or
worse a never-sell policy would have done than the strategy's own sell
logic actually did. A strategy with no AbB portfolio yet (no cycle has
stepped it since issue #42 shipped) just shows its normal row alone,
with the delta column reading as unknown rather than crashing the
command.
**Default start date and windowing (issue #57).** Without `--start`, this
windows to `config/reporting.yaml`'s `default_report_start_date` onward
(unset means unchanged, full-history behaviour); `--since-creation` opts
back into full history; an explicit `--start` always wins over both.
UNLIKE `trader performance`, windowing here does NOT re-run the
simulator — see `_sim_portfolio_metrics`'s docstring for the exact
design decision. In short: `EQUITY` and `RETURN %` are recomputed from
the WINDOW's own first/last `sim_equity_snapshots` row (mirroring
`trader performance`'s own `account.start_at`/`end_at`/`return_pct`
shape) rather than from the portfolio's live current mark-to-market, and
`WINS/LOSSES`/`STOP EXITS`/`MAX DRAWDOWN %` are all computed only from
data that falls inside the window. A window containing fewer than 2
equity snapshots shows MISSING (`-`) for `EQUITY`/`RETURN %`/`RETURN %
EXCL. STOPS` rather than a fabricated 0.00%.
"""
config = _load_config()
reporting_config = build_reporting_config()
window_start = resolve_report_start(
explicit_start=start.replace(tzinfo=UTC) if start is not None else None,
since_creation=since_creation,
default_start_date=reporting_config.default_report_start_date,
)
window_end = end.replace(tzinfo=UTC) if end is not None else None
repository = build_simulation_repository(config)
try:
portfolios = repository.all_portfolios()
except Exception as exc: # noqa: BLE001 - a stale or missing schema is common
raise _fail(
f"Could not read simulated portfolios ({_database_reason(exc)}). "
"If the database is missing or predates this version, run "
"'trader migrate'."
) from exc
if not portfolios:
typer.echo("No simulated portfolios yet — run a cycle to seed one.")
return
order, pairs = _pair_sim_portfolios(portfolios)
headers = [
"STRATEGY",
"EQUITY",
"RETURN %",
"RETURN % EXCL. STOPS",
"MAX DRAWDOWN %",
"WINS/LOSSES",
"STOP EXITS",
"ABB DELTA % (VS NORMAL)",
]
rows = _sim_report_rows(repository, order, pairs, window_start, window_end)
typer.echo(build_table(headers, rows))
def _pair_sim_portfolios(
portfolios: list[SimPortfolio],
) -> tuple[list[str], dict[str, dict[str, SimPortfolio]]]:
"""Pair each base strategy id with its (optional) AbB sibling, preserving
the order each base id was first encountered in `portfolios` — grouping
never depends on the two rows arriving adjacent to or in any particular
order relative to each other (issue #57 follow-up, finding 2's complexity
split)."""
order: list[str] = []
pairs: dict[str, dict[str, SimPortfolio]] = {}
for portfolio in portfolios:
if portfolio.strategy_id.endswith(ABB_PORTFOLIO_SUFFIX):
base_id = portfolio.strategy_id[: -len(ABB_PORTFOLIO_SUFFIX)]
slot = "abb"
else:
base_id = portfolio.strategy_id
slot = "normal"
if base_id not in pairs:
pairs[base_id] = {}
order.append(base_id)
pairs[base_id][slot] = portfolio
return order, pairs
def _sim_report_rows(
repository: SimulationRepository,
order: list[str],
pairs: dict[str, dict[str, SimPortfolio]],
window_start: datetime | None,
window_end: datetime | None,
) -> list[list[object]]:
"""One row per strategy, plus (when present) its AbB sibling row directly
beneath it carrying the "ABB DELTA %" comparison — see `sim_report`'s
docstring (issue #57 follow-up, finding 2's complexity split)."""
rows: list[list[object]] = []
for base_id in order:
slots = pairs[base_id]
normal = slots.get("normal")
abb = slots.get("abb")
normal_metrics = (
_sim_portfolio_metrics(repository, normal, window_start, window_end)
if normal is not None
else None
)
abb_metrics = (
_sim_portfolio_metrics(repository, abb, window_start, window_end)
if abb is not None
else None
)
if normal_metrics is not None:
rows.append(_sim_portfolio_row(base_id, normal_metrics, MISSING))
if abb_metrics is not None:
delta = (
_fmt_pct(abb_metrics.return_pct - normal_metrics.return_pct)
if normal_metrics is not None
and normal_metrics.return_pct is not None
and abb_metrics.return_pct is not None
else MISSING
)
rows.append(
_sim_portfolio_row(f" └ {base_id} (AbB, stop-only)", abb_metrics, delta)
)
return rows
[docs]
@app.command()
def chat(
message: str | None = typer.Option(
None,
"--message",
"-m",
help="Ask one question and print the answer, instead of an interactive loop.",
),
) -> None:
"""Chat with the local model about positions, trades, decisions, backtests,
and strategy configs (requirements §13; issues #13-#16).
A second entry point into the Core Service Layer (requirements §4): this
reads the same `trades`/`decisions`/`account_snapshots` tables the
trading loop writes and the same `config/strategies.yaml`, but needs
neither Alpaca credentials nor a running daemon — and conversely, `trader
run`/`run-once` need no chat process alive. Nothing reachable from here
can place an order or write to `config/strategies.yaml`; a backtest
request against a non-backtestable strategy (`ollama_news`) is refused
rather than attempted.
"""
try:
service = build_chat_service()
except TraderError as exc:
raise _fail(str(exc)) from exc
if message is not None:
typer.echo(service.handle(message))
return
typer.echo(
"Chat with the local model. Type 'exit' or 'quit' to leave "
"(Ctrl-C/Ctrl-D also work).\n"
)
while True:
try:
line = typer.prompt("you", prompt_suffix="> ")
except (typer.Abort, EOFError):
typer.echo("\nBye.")
break
stripped = line.strip()
if stripped.lower() in {"exit", "quit"}:
typer.echo("Bye.")
break
if not stripped:
continue
typer.echo(f"assistant> {service.handle(line)}\n")
if __name__ == "__main__": # pragma: no cover
app()