"""Build strategy instances from configuration.
The registry is the extension point: adding a strategy is one class plus one
entry here, so the set of strategies stays a configuration concern rather than
a code-structure one.
"""
from collections.abc import Sequence
from dataclasses import fields, is_dataclass
from typing import get_type_hints
from trader.config.schema import StrategyConfig
from trader.errors import ConfigError
from trader.strategies.alligator import AlligatorStrategy
from trader.strategies.base import Strategy
from trader.strategies.bollinger_revert import BollingerRevertStrategy
from trader.strategies.llm_strategy import LlmStrategy
from trader.strategies.ma_crossover import MaCrossoverStrategy
from trader.strategies.macd import MacdStrategy
from trader.strategies.resistance_breakout import ResistanceBreakoutStrategy
from trader.strategies.rsi_revert import RsiRevertStrategy
from trader.strategies.trailing_stop_floor import TrailingStopFloorStrategy
from trader.strategies.turtle import TurtleStrategy
__all__ = ["STRATEGY_TYPES", "build_all", "build_strategy"]
STRATEGY_TYPES: dict[str, type] = {
"rsi_revert": RsiRevertStrategy,
"bollinger_revert": BollingerRevertStrategy,
"turtle": TurtleStrategy,
"ma_crossover": MaCrossoverStrategy,
"alligator": AlligatorStrategy,
"macd": MacdStrategy,
# issue #36: a validated-band alternative to `turtle`'s naive N-day-high
# channel. Its Decimal-valued params (`region_width_pct`,
# `breakout_buffer_pct`) need no special registry handling -- unlike
# `llm`/`trailing_stop_floor`, this strategy needs no injected
# collaborator, and its own `__post_init__` (not this registry) converts
# and validates them, the same division of labour `TurtleStrategy`'s
# plain `int` fields already get from the generic path below.
"resistance_breakout": ResistanceBreakoutStrategy,
}
#: Parameters an `llm` entry may carry that configure the *provider* rather
#: than the strategy. The caller builds the provider, so the registry accepts
#: and ignores them rather than rejecting a valid config file.
_PROVIDER_PARAMS = frozenset({"model", "temperature", "timeout_seconds", "num_ctx"})
#: Parameters `LlmStrategy` itself accepts.
#:
#: Deliberately excludes `clock`, which `LlmStrategy.__init__` also accepts:
#: it exists only so a test can inject a frozen time, and it must never
#: become reachable from YAML. A config-settable clock would either freeze
#: the recency window at one instant forever, or — given a non-callable
#: value like a plain YAML string — raise `TypeError: 'str' object is not
#: callable` where `now = self._clock()` is read, outside any `try`,
#: aborting the symbol loop before the protection ratchet runs. See
#: `tests/strategies/test_registry_llm.py`.
#:
#: Also deliberately excludes `decision_memory`, for the same reason as `clock`
#: and for the same reason the safety gate is not a flag: it is a collaborator
#: the CLI wires from the configured database, not a value, and a YAML string
#: reaching it would raise `AttributeError: 'str' object has no attribute
#: 'latest_decision'` inside `_reuse_or_none` — which is caught there, so the
#: symptom would be a gate that silently never fires rather than a startup
#: failure. The two *numbers* that tune the gate are settable; the wiring is not.
#:
#: `alias_resolver` is excluded on the same grounds: it is a callable the CLI
#: builds from the broker's asset lookup, and a YAML string reaching it would
#: raise `TypeError: 'str' object is not callable` inside `_aliases_for` —
#: caught there, so the symptom would be relevance silently falling back to the
#: bare ticker, which is exactly the issue #6 bug it exists to fix.
#: `company_aliases` stays settable, because that one *is* a value: an operator
#: naming "Google" for GOOG supplies knowledge the asset record lacks.
#:
#: `analyst_provider` is excluded on the same grounds as `alias_resolver`: it
#: is the same `AnalystProvider` discovery already builds via
#: `build_analyst_provider()`, wired here rather than named in YAML, and a
#: YAML string reaching it would raise `AttributeError: 'str' object has no
#: attribute 'get_opinion'` inside `_fetch_analyst_opinion` — caught there,
#: so the symptom would be analyst data silently never reaching the prompt.
#:
#: `earnings_provider` is excluded on the identical grounds, one issue later
#: (issue #89): it is the same `EarningsProvider` the CLI builds via its own
#: `build_earnings_provider()`, wired here rather than named in YAML, and a
#: YAML string reaching it would raise `AttributeError: 'str' object has no
#: attribute 'get_next_earnings_date'` inside `_fetch_earnings_date` —
#: caught there too, so the symptom is the same silent "this data never
#: reaches the prompt".
_LLM_STRATEGY_PARAMS = frozenset(
{
"max_news_items",
"max_news_age_hours",
"max_reuse_age_minutes",
"reuse_price_band_pct",
"lookback_bars",
"company_aliases",
}
)
#: Parameters `TrailingStopFloorStrategy` itself accepts (requirements
#: 6.1/6.3, issue #18). Built the same way as `llm`, below: it needs an
#: `LlmProvider` collaborator the CLI builds from the entry's own
#: `model`/`temperature`/`timeout_seconds`/`num_ctx` (`_PROVIDER_PARAMS`,
#: shared with `llm`), so it cannot be a plain dataclass entry in
#: `STRATEGY_TYPES` — that path calls `strategy_class(id=..., **config.params)`
#: with no collaborator injection at all, which is exactly why `llm` needed
#: its own branch in `build_strategy` in the first place.
_TRAILING_STOP_FLOOR_PARAMS = frozenset(
{
"tickers",
"trail_pct",
"trail_amount",
"trail_min",
"trail_max",
"floor_price",
"floor_pct",
"notes_path",
"lookback_bars",
}
)
[docs]
def build_strategy(
config: StrategyConfig,
*,
llm: object | None = None,
news_provider: object | None = None,
decision_memory: object | None = None,
alias_resolver: object | None = None,
analyst_provider: object | None = None,
earnings_provider: object | None = None,
) -> Strategy:
"""Instantiate one configured strategy.
`decision_memory` is optional and only ever reaches an `llm` entry: a
strategy built without one calls the model every cycle, which is the
behaviour that existed before the news-fingerprint gate. Passing `None` is a
valid, conservative configuration, not a degraded one.
Raises:
ConfigError: unknown type, a parameter the strategy does not accept, or
an `llm`/`trailing_stop_floor` entry with no provider supplied.
"""
if config.type == "llm":
return _build_llm_strategy(
config,
llm,
news_provider,
decision_memory,
alias_resolver,
analyst_provider,
earnings_provider,
)
if config.type == "trailing_stop_floor":
return _build_trailing_stop_floor_strategy(config, llm)
strategy_class = STRATEGY_TYPES.get(config.type)
if strategy_class is None:
# "llm" and "trailing_stop_floor" aren't in STRATEGY_TYPES (both have
# their own construction path above, for the same reason: a
# collaborator the CLI must inject, which the generic path below has
# no way to pass), but a typo like "llmm" should still see them as
# options.
valid = ", ".join(sorted([*STRATEGY_TYPES, "llm", "trailing_stop_floor"]))
raise ConfigError(
f"Unknown strategy type {config.type!r} for id {config.id!r}. "
f"Valid types: {valid}."
)
if not is_dataclass(strategy_class):
raise ConfigError(
f"Strategy type {config.type!r} maps to {strategy_class.__name__}, "
"which is not a dataclass. The registry reads accepted parameters "
"from dataclass fields, so strategy classes must be dataclasses."
)
accepted = {f.name for f in fields(strategy_class)} - {"id"}
unknown = set(config.params) - accepted
if unknown:
raise ConfigError(
f"Strategy {config.id!r} ({config.type}) got unknown parameters: "
f"{', '.join(sorted(unknown))}. Accepted: {', '.join(sorted(accepted))}."
)
hints = get_type_hints(strategy_class)
for name, value in config.params.items():
expected = hints.get(name)
if expected is int:
# bool is an int subclass; a YAML `true` is not a period.
if isinstance(value, bool) or not isinstance(value, int):
raise ConfigError(
f"Strategy {config.id!r} parameter {name!r} must be an "
f"integer, got {value!r} ({type(value).__name__})."
)
# A lookback period below 1 is a config typo, but it surfaces deep
# in the indicators as a bare `ValueError` that names neither the
# file nor the strategy. Catch it where the file is still in hand.
if value < 1 and name.endswith("period"):
raise ConfigError(
f"Strategy {config.id!r} parameter {name!r} must be >= 1, "
f"got {value}."
)
elif expected is float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ConfigError(
f"Strategy {config.id!r} parameter {name!r} must be a "
f"number, got {value!r} ({type(value).__name__})."
)
return strategy_class(id=config.id, **config.params)
def _build_llm_strategy(
config: StrategyConfig,
llm: object | None,
news_provider: object | None,
decision_memory: object | None = None,
alias_resolver: object | None = None,
analyst_provider: object | None = None,
earnings_provider: object | None = None,
) -> Strategy:
"""Build an `LlmStrategy`, which needs collaborators rather than only params.
Handled separately because the generic path reads accepted parameters from
dataclass fields, and `LlmStrategy` is not a dataclass — it holds a model
client and a news provider, and records what its last call saw.
"""
if llm is None or news_provider is None:
raise ConfigError(
f"Strategy {config.id!r} has type 'llm' but no model or news "
"provider was supplied. Build it through the CLI, which wires "
"both from configuration."
)
unknown = set(config.params) - _LLM_STRATEGY_PARAMS - _PROVIDER_PARAMS
if unknown:
raise ConfigError(
f"Strategy {config.id!r} (llm) got unknown parameters: "
f"{', '.join(sorted(unknown))}. Accepted: "
f"{', '.join(sorted(_LLM_STRATEGY_PARAMS | _PROVIDER_PARAMS))}."
)
kwargs = {k: v for k, v in config.params.items() if k in _LLM_STRATEGY_PARAMS}
aliases = kwargs.get("company_aliases")
if aliases is not None:
kwargs["company_aliases"] = tuple(aliases)
return LlmStrategy(
id=config.id,
llm=llm,
news_provider=news_provider,
decision_memory=decision_memory,
alias_resolver=alias_resolver,
analyst_provider=analyst_provider,
earnings_provider=earnings_provider,
**kwargs,
)
def _build_trailing_stop_floor_strategy(
config: StrategyConfig, llm: object | None
) -> Strategy:
"""Build a `TrailingStopFloorStrategy`, which needs an LLM collaborator
the same way `llm` does, and for the same reason — see that function.
`news_provider`/`decision_memory`/`alias_resolver` are not accepted here:
this strategy reads no news feed, has no reuse gate to key off a
decision's fingerprint, and computes no relevance score, so there is
nothing for those three collaborators to do.
"""
if llm is None:
raise ConfigError(
f"Strategy {config.id!r} has type 'trailing_stop_floor' but no "
"model was supplied. Build it through the CLI, which wires one "
"from the entry's own model/temperature/timeout_seconds/num_ctx "
"params."
)
unknown = set(config.params) - _TRAILING_STOP_FLOOR_PARAMS - _PROVIDER_PARAMS
if unknown:
raise ConfigError(
f"Strategy {config.id!r} (trailing_stop_floor) got unknown "
f"parameters: {', '.join(sorted(unknown))}. Accepted: "
f"{', '.join(sorted(_TRAILING_STOP_FLOOR_PARAMS | _PROVIDER_PARAMS))}."
)
kwargs = {k: v for k, v in config.params.items() if k in _TRAILING_STOP_FLOOR_PARAMS}
return TrailingStopFloorStrategy(id=config.id, llm=llm, **kwargs)
[docs]
def build_all(
configs: Sequence[StrategyConfig],
*,
llm: object | None = None,
news_provider: object | None = None,
decision_memory: object | None = None,
) -> dict[str, Strategy]:
"""Instantiate every configured strategy, keyed by id."""
return {
c.id: build_strategy(
c, llm=llm, news_provider=news_provider, decision_memory=decision_memory
)
for c in configs
}