"""Turning a chat message into what to look up before answering it.
Deterministic keyword/regex classification — never a model call. The hard
rule that a conversational backtest trigger must never route to
`LlmStrategy` (`backtestable = False`) has to hold structurally, and a rule
engine is exhaustively testable in the way an LLM classification would not
be. This module only decides **what data to fetch**; `ChatService` still asks
the model to compose the natural-language answer once that data is in hand,
the same two-step shape `LlmStrategy` uses (build inputs, then ask).
A simple heuristic, not NLU: `classify()` never raises, an unrecognised
message always falls back to `Intent.GENERAL`, and every extracted field is
optional — a caller that gets `None` back asks the user to be more specific
rather than guessing.
"""
from __future__ import annotations
import re
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import date
from enum import Enum
__all__ = ["Intent", "ParsedIntent", "classify"]
[docs]
class Intent(Enum):
"""What kind of chat turn this is, and therefore what to fetch."""
GENERAL = "general"
EXPLAIN_DECISION = "explain_decision"
BACKTEST = "backtest"
STRATEGY_REVIEW = "strategy_review"
[docs]
@dataclass(frozen=True, slots=True)
class ParsedIntent:
"""What `classify()` extracted from one message. Every field but the
first two is intent-specific and may be `None` when the message did not
say."""
intent: Intent
raw_message: str
#: EXPLAIN_DECISION only.
symbol: str | None = None
decision_id: int | None = None
on_date: date | None = None
#: BACKTEST (strategy_id also used to focus STRATEGY_REVIEW).
strategy_id: str | None = None
ticker: str | None = None
start: date | None = None
end: date | None = None
_DATE_RE = re.compile(r"\b(\d{4})-(\d{2})-(\d{2})\b")
_DECISION_ID_RE = re.compile(r"\bdecision\s*(?:id\s*)?#?\s*(\d+)\b", re.IGNORECASE)
#: Tickers on major US exchanges are conventionally 1-5 letters.
_MAX_TICKER_LETTERS = 5
_BACKTEST_RE = re.compile(r"back[\s-]?test", re.IGNORECASE)
_STRATEGY_REVIEW_RE = re.compile(
r"\bstrateg(?:y|ies)\b.{0,40}\b(config|configs|yaml|review|setting|settings|param|params|parameter)",
re.IGNORECASE,
)
_EXPLAIN_DECISION_RE = re.compile(
r"\bdecision\b"
r"|\bwhy\s+(?:did|do|does|would)\s+"
r"(?:you|it|the\s+(?:app|bot|model|strategy))\b",
re.IGNORECASE,
)
#: Short, common English words that would otherwise be picked up as ticker
#: candidates by `_find_ticker`'s 1-5 letter scan. Not exhaustive — this is a
#: v1 heuristic, not a real NER model — but it covers the words that
#: actually turn up in the phrasing this app's own tests and a person
#: typing casually both produce.
_TICKER_STOPWORDS = frozenset(
{
"I",
"A",
"AN",
"ON",
"OF",
"TO",
"IN",
"AT",
"IS",
"IT",
"MY",
"THE",
"AND",
"OR",
"WITH",
"FROM",
"BETWEEN",
"VS",
"DID",
"WHY",
"HOW",
"WHAT",
"WHEN",
"WHO",
"ME",
"DO",
"DOES",
"YOU",
"FOR",
"ABOUT",
"OK",
"OKAY",
"SO",
"BE",
"AS",
"IF",
"BY",
"UP",
"NO",
"GO",
"ARE",
"WAS",
"WERE",
"HAS",
"HAVE",
"HAD",
"NOT",
"THAT",
"THIS",
"THAN",
"THEN",
"OUR",
"US",
"CAN",
"ALL",
"ANY",
"RUN",
"GET",
"SEE",
"OUT",
"OFF",
"ID",
}
)
def _find_dates(message: str) -> list[date]:
"""Every ISO `YYYY-MM-DD` date in the message, in the order they appear."""
found: list[date] = []
for match in _DATE_RE.finditer(message):
try:
found.append(
date(int(match.group(1)), int(match.group(2)), int(match.group(3)))
)
except ValueError:
# e.g. 2026-13-40 — a malformed date is not a date. Skipped, not
# raised: a chat classifier degrading a bad date to "no date
# found" is the safe direction, same as `parse_decision`'s "an
# unusable value is a missing value" discipline.
continue
return found
def _strip_strategy_id(text: str, strategy_id: str | None) -> str:
"""Remove a matched strategy id from `text` before scanning for a ticker.
Without this, a strategy id like `rsi_revert_14` contributes the token
"rsi" (3 letters, not a stopword) to `_find_ticker`'s scan and gets
mistaken for the ticker itself. `_` is tolerant of the id being typed
with spaces or hyphens instead of underscores, matching how
`_find_strategy_id` locates it in the first place.
"""
if not strategy_id:
return text
pattern = re.compile(re.escape(strategy_id).replace("_", r"[\s_-]+"), re.IGNORECASE)
return pattern.sub(" ", text)
def _find_ticker(message: str, *, exclude: Sequence[str] = ()) -> str | None:
"""The first token that looks like a ticker: 1-5 letters, not a stopword.
Tokenizes on whole alphabetic runs (`[A-Za-z]+`) and *then* filters by
length, rather than matching `{1,5}` directly — the latter chops a long
word like "backtest" into "BACKT" + "EST" instead of excluding it, since
a bounded quantifier with no word boundary still matches mid-word.
"""
excluded = {e.upper() for e in exclude}
for token in re.findall(r"[A-Za-z]+", message):
if not 1 <= len(token) <= _MAX_TICKER_LETTERS:
continue
upper = token.upper()
if upper in _TICKER_STOPWORDS or upper in excluded:
continue
return upper
return None
def _find_strategy_id(message: str, strategy_ids: Sequence[str]) -> str | None:
"""A configured strategy id mentioned in the message, if any.
Matches with spaces/hyphens treated as underscores, so "rsi revert 14"
and "rsi_revert_14" both hit the configured id `rsi_revert_14`. When more
than one configured id appears, the longest wins — a shorter id that
happens to be a substring of a longer one (there is no such pair
configured today, but nothing stops one existing) must not shadow it.
"""
normalized = re.sub(r"[\s-]+", "_", message.lower())
matches = [sid for sid in strategy_ids if sid.lower() in normalized]
if not matches:
return None
return max(matches, key=len)
[docs]
def classify(message: str, *, strategy_ids: Sequence[str] = ()) -> ParsedIntent:
"""Route one chat message. Never raises.
`strategy_ids` is the configured strategies' ids (`config/strategies.yaml`),
used only to recognise one by name inside a BACKTEST or STRATEGY_REVIEW
message — it never changes which `Intent` a message is classified as.
"""
text = message or ""
if _BACKTEST_RE.search(text):
strategy_id = _find_strategy_id(text, strategy_ids)
dates = _find_dates(text)
ticker = _find_ticker(_strip_strategy_id(text, strategy_id))
return ParsedIntent(
intent=Intent.BACKTEST,
raw_message=text,
strategy_id=strategy_id,
ticker=ticker,
start=dates[0] if dates else None,
end=dates[1] if len(dates) > 1 else None,
)
if _EXPLAIN_DECISION_RE.search(text) and not _STRATEGY_REVIEW_RE.search(text):
decision_match = _DECISION_ID_RE.search(text)
dates = _find_dates(text)
return ParsedIntent(
intent=Intent.EXPLAIN_DECISION,
raw_message=text,
symbol=_find_ticker(text),
decision_id=int(decision_match.group(1)) if decision_match else None,
on_date=dates[0] if dates else None,
)
if _STRATEGY_REVIEW_RE.search(text):
return ParsedIntent(
intent=Intent.STRATEGY_REVIEW,
raw_message=text,
strategy_id=_find_strategy_id(text, strategy_ids),
)
return ParsedIntent(intent=Intent.GENERAL, raw_message=text)