feat: fxhnt foundation — hexagonal architecture + proven gauntlet + vertical slice

Enterprise clean-rebuild (no foxhunt code). Hexagonal/ports-and-adapters: pure domain (gauntlet
math, strategies, backtest, models) | ports (DataProvider, repositories) | adapters (Yahoo data,
SQLAlchemy operational [Postgres/SQLite], DuckDB analytical) | application (ResearchService, DI) |
CLI composition root. Gauntlet-first: Deflated Sharpe (Bailey-LdP) built + falsification-tested
(kills best-of-N-on-noise, keeps real premium). Full vertical slice runs end-to-end on real data:
data -> strategy(trend) -> backtest(net of costs) -> IS/OOS gauntlet -> persistence. 4/4 tests green.
Postgres+DuckDB split, pydantic contracts, typed, DRY via one-contract-per-port. ADR + README.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,11 @@
+__pycache__/
+*.pyc
+*.egg-info/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+*.duckdb
+*.db
+.env
+build/
+dist/

README.md

@@ -0,0 +1,38 @@
+# fxhnt
+
+Agentic strategy-research & multi-strategy execution platform. It systematically **discovers,
+backtests and out-of-sample-validates** trading strategies across many markets, keeps only what
+survives a rigorous statistical gauntlet, and runs the survivors live (multiple strategies at once).
+
+The bet is not a secret edge — it's **breadth + discipline + automation**. The hard part (and the
+moat) is refusing to fool yourself at scale; the validation gauntlet is the core, built and proven
+first.
+
+## Architecture (hexagonal / ports-and-adapters)
+```
+src/fxhnt/
+  domain/        pure logic: gauntlet (Deflated Sharpe), strategies, backtest, models   ← no I/O
+  ports/         contracts: DataProvider, repositories (the only seams)
+  adapters/      infra: yahoo data, SQLAlchemy (Postgres/SQLite) + DuckDB stores
+  application/   use-case services (ResearchService) — orchestrate via ports
+  cli.py         composition root (wires concrete adapters)
+```
+See `docs/architecture/0001-architecture.md`.
+
+## Quickstart
+```bash
+pip install -e ".[dev]"
+
+pytest                                   # unit (gauntlet falsification) + integration (vertical slice)
+fxhnt strategies                         # list strategy kinds
+fxhnt research SPY --kind trend --window 200    # data → backtest → gauntlet → persist
+fxhnt list --passed-only                 # the survivor library
+```
+
+Config via `FXHNT_*` env vars (e.g. `FXHNT_OPERATIONAL_DSN=postgresql+psycopg://...`). Defaults to
+SQLite + a local DuckDB file under `~/.fxhnt/`.
+
+## Status
+Vertical slice working: data (Yahoo) → strategy (trend) → backtest (net of costs) → IS/OOS gauntlet
+→ persistence (operational + analytical). Next: the multi-strategy execution layer, more strategy
+templates + data adapters, and the agentic discovery search on top of the proven gauntlet.

docs/architecture/0001-architecture.md

@@ -0,0 +1,44 @@
+# ADR 0001 — Foundational architecture
+
+**Status:** accepted · **Date:** 2026-06-09
+
+## Context
+fxhnt is an agentic platform that systematically discovers, backtests and OOS-validates trading
+strategies across many markets — and runs the survivors live (multiple strategies at once). The value
+is NOT a secret edge; it is **breadth + discipline + automation**. The #1 existential risk is
+multiple-testing: a fast search over thousands of (strategy × market) combos manufactures false
+positives unless the statistics correct for the full search. This must be designed-in, not bolted-on.
+
+## Decisions
+
+1. **Hexagonal / ports-and-adapters (clean architecture).**
+   - `domain/` — pure business logic (gauntlet math, strategies, backtest, portfolio). No I/O.
+   - `ports/` — abstract contracts (DataProvider, Broker, repositories).
+   - `adapters/` — concrete infra (Yahoo/Databento data, IBKR broker, SQLAlchemy/DuckDB stores).
+   - `application/` — use-case services that orchestrate the domain via ports (dependency-injected).
+   - `cli.py` — the single composition root that wires concrete adapters.
+   - Rationale: swappable infra, isolated testability, DRY (one contract per port, no copy-paste adapters).
+
+2. **Gauntlet first.** The Deflated-Sharpe validation engine (Bailey & López de Prado) is built and
+   *falsification-tested* (must kill best-of-N-on-noise, keep a real premium) BEFORE any search is
+   built on top. `n_trials`/`sr_variance` carry the full search size into the verdict.
+
+3. **Persistence split: Postgres (operational) + DuckDB (analytical).** Relational store for runs,
+   verdicts, the survivor library, positions, trades; columnar store for market data + backtest
+   timeseries. Both behind repository ports; SQLAlchemy makes the operational store DB-agnostic
+   (SQLite for dev/test, Postgres in production via `FXHNT_OPERATIONAL_DSN`).
+
+4. **Contracts via pydantic; numeric value objects via frozen dataclasses.** Serializable, validated
+   models cross boundaries; numpy-holding objects (PriceSeries, BacktestResult) stay in-memory.
+
+5. **Clean rebuild (no foxhunt code).** A pristine codebase; proven ideas are re-implemented cleanly.
+
+## Non-negotiable principles
+- `domain/` imports nothing infrastructural (enforced by review/structure).
+- Every strategy must declare a structural `rationale` to register (the gauntlet requires one).
+- Conservative validation defaults (DSR ≥ 0.95 over the full search; OOS must hold).
+
+## Build order (so we never ship a POC pretending to be an app)
+gauntlet → contracts/domain → data adapter → persistence → application slice → **execution layer
+(multi-strategy)** → discovery/agentic search → portfolio assembly → live execution.
+This pass delivers a full vertical slice (data → strategy → backtest → gauntlet → persistence).

pyproject.toml

@@ -0,0 +1,41 @@
+[project]
+name = "fxhnt"
+version = "0.1.0"
+description = "Agentic strategy-research & multi-strategy execution platform — discover, backtest, OOS-validate at scale"
+requires-python = ">=3.11"
+dependencies = [
+    "numpy>=1.26",
+    "pydantic>=2.6",
+    "pydantic-settings>=2.2",
+    "sqlalchemy>=2.0",
+    "duckdb>=1.0",
+    "typer>=0.12",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8.0", "hypothesis>=6.100", "ruff>=0.5", "mypy>=1.10"]
+
+[project.scripts]
+fxhnt = "fxhnt.cli:app"
+
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.ruff]
+line-length = 120
+target-version = "py311"
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "SIM"]
+
+[tool.mypy]
+python_version = "3.11"
+strict = true
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+testpaths = ["tests"]

src/fxhnt/__init__.py

@@ -0,0 +1,7 @@
+"""fxhnt — agentic strategy-research & multi-strategy execution platform.
+
+Layered (hexagonal / ports-and-adapters) so the domain logic stays pure and the infrastructure
+(data sources, brokers, databases) is swappable behind contracts. See docs/architecture/.
+"""
+
+__version__ = "0.1.0"

src/fxhnt/adapters/__init__.py

@@ -0,0 +1 @@
+"""Adapters — concrete implementations of the ports (infrastructure). The domain never imports these."""

src/fxhnt/adapters/data/__init__.py

@@ -0,0 +1,3 @@
+from fxhnt.adapters.data.yahoo import YahooDataProvider
+
+__all__ = ["YahooDataProvider"]

src/fxhnt/adapters/data/yahoo.py

@@ -0,0 +1,48 @@
+"""Yahoo Finance adapter — free daily adjusted closes. Implements the DataProvider port.
+
+Bakes in a hard-won lesson: drop today's INCOMPLETE intraday bar (it corrupts vol/trend signals
+when the platform runs mid-session). Strategies here are daily-CLOSE strategies.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import urllib.request
+
+import numpy as np
+
+from fxhnt.domain.models import Market, PriceSeries
+
+_BASE = "https://query1.finance.yahoo.com/v8/finance/chart"
+
+
+class YahooDataProvider:
+    name = "yahoo"
+
+    def __init__(self, default_range: str = "25y", timeout: int = 30) -> None:
+        self._range = default_range
+        self._timeout = timeout
+
+    def fetch(self, market: Market, start: str | None = None, end: str | None = None) -> PriceSeries:
+        url = f"{_BASE}/{market.symbol}?interval=1d&range={self._range}"
+        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+        payload = json.loads(urllib.request.urlopen(req, timeout=self._timeout).read())
+        res = payload["chart"]["result"][0]
+        ts = res["timestamp"]
+        ind = res["indicators"]
+        adj = ind.get("adjclose", [{}])[0].get("adjclose") or ind["quote"][0]["close"]
+        today = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d")
+        rows: list[tuple[str, float]] = []
+        for t, c in zip(ts, adj):
+            if c is None:
+                continue
+            d = dt.datetime.fromtimestamp(t, dt.timezone.utc).strftime("%Y-%m-%d")
+            if d == today:            # incomplete intraday bar
+                continue
+            if start and d < start:
+                continue
+            if end and d > end:
+                continue
+            rows.append((d, float(c)))
+        return PriceSeries(market=market, dates=tuple(d for d, _ in rows),
+                           close=np.array([c for _, c in rows], dtype=float))

src/fxhnt/adapters/persistence/__init__.py

@@ -0,0 +1,4 @@
+from fxhnt.adapters.persistence.analytical import DuckDbAnalyticalStore
+from fxhnt.adapters.persistence.operational import SqlOperationalRepository
+
+__all__ = ["DuckDbAnalyticalStore", "SqlOperationalRepository"]

src/fxhnt/adapters/persistence/analytical.py

@@ -0,0 +1,55 @@
+"""DuckDB analytical store — columnar timeseries (market prices + backtest returns). Implements
+AnalyticalStore. Embedded file, no server; connect-per-operation (simple + safe for batch research)."""
+from __future__ import annotations
+
+import duckdb
+import numpy as np
+
+from fxhnt.domain.models import Market, PriceSeries
+
+_SCHEMA = [
+    "CREATE TABLE IF NOT EXISTS prices (symbol VARCHAR, asset_class VARCHAR, date VARCHAR, close DOUBLE)",
+    "CREATE TABLE IF NOT EXISTS returns (run_id VARCHAR, date VARCHAR, ret DOUBLE)",
+]
+
+
+class DuckDbAnalyticalStore:
+    def __init__(self, path: str) -> None:
+        self._path = path
+        con = duckdb.connect(path)
+        for ddl in _SCHEMA:
+            con.execute(ddl)
+        con.close()
+
+    def save_prices(self, prices: PriceSeries) -> None:
+        m = prices.market
+        rows = [(m.symbol, m.asset_class.value, d, float(c)) for d, c in zip(prices.dates, prices.close)]
+        con = duckdb.connect(self._path)
+        try:
+            con.execute("DELETE FROM prices WHERE symbol = ? AND asset_class = ?", [m.symbol, m.asset_class.value])
+            con.executemany("INSERT INTO prices VALUES (?, ?, ?, ?)", rows)
+        finally:
+            con.close()
+
+    def load_prices(self, market: Market) -> PriceSeries | None:
+        con = duckdb.connect(self._path)
+        try:
+            rows = con.execute(
+                "SELECT date, close FROM prices WHERE symbol = ? AND asset_class = ? ORDER BY date",
+                [market.symbol, market.asset_class.value],
+            ).fetchall()
+        finally:
+            con.close()
+        if not rows:
+            return None
+        return PriceSeries(market=market, dates=tuple(r[0] for r in rows),
+                           close=np.array([r[1] for r in rows], dtype=float))
+
+    def save_returns(self, run_id: str, dates: tuple[str, ...], returns: np.ndarray) -> None:
+        rows = [(run_id, d, float(r)) for d, r in zip(dates, returns)]
+        con = duckdb.connect(self._path)
+        try:
+            con.execute("DELETE FROM returns WHERE run_id = ?", [run_id])
+            con.executemany("INSERT INTO returns VALUES (?, ?, ?)", rows)
+        finally:
+            con.close()

src/fxhnt/adapters/persistence/operational.py

@@ -0,0 +1,55 @@
+"""SQL operational repository — implements OperationalRepository over SQLAlchemy (Postgres or SQLite).
+Maps the domain ResearchRun <-> the ORM row; the domain never sees SQLAlchemy."""
+from __future__ import annotations
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from fxhnt.adapters.persistence.sql_models import Base, ResearchRunRow
+from fxhnt.domain.models import AssetClass, BacktestStats, Market, ResearchRun, StrategySpec, Verdict
+
+
+class SqlOperationalRepository:
+    def __init__(self, dsn: str) -> None:
+        self._engine = create_engine(dsn, future=True)
+        Base.metadata.create_all(self._engine)
+
+    @staticmethod
+    def _to_row(run: ResearchRun) -> ResearchRunRow:
+        return ResearchRunRow(
+            run_id=run.run_id, symbol=run.market.symbol, asset_class=run.market.asset_class.value,
+            venue=run.market.venue, currency=run.market.currency, strategy_kind=run.spec.kind,
+            params=dict(run.spec.params), cagr=run.stats.cagr, ann_vol=run.stats.ann_vol,
+            sharpe=run.stats.sharpe, max_drawdown=run.stats.max_drawdown, n_obs=run.stats.n_obs,
+            passed=run.verdict.passed, dsr=run.verdict.dsr, is_sharpe=run.verdict.is_sharpe,
+            oos_sharpe=run.verdict.oos_sharpe, n_trials=run.n_trials, reasons=list(run.verdict.reasons),
+            created_at=run.created_at,
+        )
+
+    @staticmethod
+    def _to_domain(row: ResearchRunRow) -> ResearchRun:
+        return ResearchRun(
+            run_id=row.run_id,
+            market=Market(symbol=row.symbol, asset_class=AssetClass(row.asset_class), venue=row.venue, currency=row.currency),
+            spec=StrategySpec(kind=row.strategy_kind, params=row.params),
+            stats=BacktestStats(cagr=row.cagr, ann_vol=row.ann_vol, sharpe=row.sharpe, max_drawdown=row.max_drawdown, n_obs=row.n_obs),
+            verdict=Verdict(passed=row.passed, dsr=row.dsr, is_sharpe=row.is_sharpe, oos_sharpe=row.oos_sharpe, n_trials=row.n_trials, reasons=row.reasons),
+            n_trials=row.n_trials, created_at=row.created_at,
+        )
+
+    def save_run(self, run: ResearchRun) -> None:
+        with Session(self._engine) as s:
+            s.merge(self._to_row(run))
+            s.commit()
+
+    def get_run(self, run_id: str) -> ResearchRun | None:
+        with Session(self._engine) as s:
+            row = s.get(ResearchRunRow, run_id)
+            return self._to_domain(row) if row else None
+
+    def list_runs(self, *, passed_only: bool = False) -> list[ResearchRun]:
+        stmt = select(ResearchRunRow).order_by(ResearchRunRow.created_at.desc())
+        if passed_only:
+            stmt = stmt.where(ResearchRunRow.passed.is_(True))
+        with Session(self._engine) as s:
+            return [self._to_domain(r) for r in s.scalars(stmt)]

src/fxhnt/adapters/persistence/sql_models.py

@@ -0,0 +1,36 @@
+"""SQLAlchemy 2.0 ORM models for the operational store. JSON columns work on both Postgres and SQLite,
+so the same models serve production (Postgres) and dev/test (SQLite)."""
+from __future__ import annotations
+
+import datetime as dt
+
+from sqlalchemy import JSON, Boolean, DateTime, Float, Integer, String
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class ResearchRunRow(Base):
+    __tablename__ = "research_runs"
+
+    run_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    symbol: Mapped[str] = mapped_column(String(32), index=True)
+    asset_class: Mapped[str] = mapped_column(String(16))
+    venue: Mapped[str] = mapped_column(String(16))
+    currency: Mapped[str] = mapped_column(String(8))
+    strategy_kind: Mapped[str] = mapped_column(String(32), index=True)
+    params: Mapped[dict] = mapped_column(JSON)
+    cagr: Mapped[float] = mapped_column(Float)
+    ann_vol: Mapped[float] = mapped_column(Float)
+    sharpe: Mapped[float] = mapped_column(Float)
+    max_drawdown: Mapped[float] = mapped_column(Float)
+    n_obs: Mapped[int] = mapped_column(Integer)
+    passed: Mapped[bool] = mapped_column(Boolean, index=True)
+    dsr: Mapped[float] = mapped_column(Float)
+    is_sharpe: Mapped[float] = mapped_column(Float)
+    oos_sharpe: Mapped[float] = mapped_column(Float)
+    n_trials: Mapped[int] = mapped_column(Integer)
+    reasons: Mapped[list] = mapped_column(JSON)
+    created_at: Mapped[dt.datetime] = mapped_column(DateTime)

src/fxhnt/application/__init__.py

@@ -0,0 +1,5 @@
+"""Application layer — use-case services that orchestrate the domain via ports. No business math here,
+no infrastructure here; just coordination."""
+from fxhnt.application.research import ResearchService
+
+__all__ = ["ResearchService"]

src/fxhnt/application/research.py

@@ -0,0 +1,66 @@
+"""Research use-case — orchestrates the vertical slice: data → backtest → IS/OOS gauntlet → persist.
+
+Depends only on PORTS (DataProvider, OperationalRepository, AnalyticalStore), injected by the caller
+(the CLI composition root). Holds no infrastructure knowledge — fully unit-testable with fakes.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import hashlib
+
+from fxhnt.config import Settings
+from fxhnt.domain.backtest import run_backtest
+from fxhnt.domain.gauntlet import evaluate
+from fxhnt.domain.models import Market, ResearchRun, StrategySpec
+from fxhnt.domain.strategies import get_strategy
+from fxhnt.ports.data import DataProvider
+from fxhnt.ports.repository import AnalyticalStore, OperationalRepository
+
+_MIN_HISTORY = 300
+
+
+def _run_id(market: Market, spec: StrategySpec) -> str:
+    h = hashlib.sha1(f"{market}|{spec.key()}".encode()).hexdigest()[:8]
+    return f"{market.symbol}-{spec.kind}-{h}"
+
+
+class ResearchService:
+    def __init__(self, data: DataProvider, operational: OperationalRepository,
+                 analytical: AnalyticalStore, settings: Settings) -> None:
+        self._data = data
+        self._op = operational
+        self._an = analytical
+        self._s = settings
+
+    def evaluate_candidate(self, market: Market, spec: StrategySpec, *,
+                           n_trials: int = 1, sr_variance: float = 0.0, persist: bool = True) -> ResearchRun:
+        # 1. data — cache-first in the analytical store, fall back to the provider
+        prices = self._an.load_prices(market)
+        if prices is None or len(prices) < _MIN_HISTORY:
+            prices = self._data.fetch(market)
+            self._an.save_prices(prices)
+
+        # 2. backtest (net of costs)
+        bt = run_backtest(prices, spec, cost_bps_per_turnover=self._s.cost_bps_per_turnover)
+
+        # 3. in-sample / out-of-sample split
+        r = bt.returns
+        split = int((1.0 - self._s.gauntlet.oos_fraction) * len(r))
+
+        # 4. the gauntlet (n_trials/sr_variance carry the FULL search size — set by the discovery layer)
+        strategy = get_strategy(spec.kind)
+        verdict = evaluate(
+            r[:split], r[split:], n_trials=n_trials, sr_variance=sr_variance,
+            dsr_min=self._s.gauntlet.dsr_min, oos_min_sharpe=self._s.gauntlet.oos_min_sharpe,
+            max_is_oos_decay=self._s.gauntlet.max_is_oos_decay,
+            has_economic_rationale=bool(getattr(strategy, "rationale", "")),
+        )
+
+        run = ResearchRun(
+            run_id=_run_id(market, spec), market=market, spec=spec, stats=bt.stats,
+            verdict=verdict, n_trials=n_trials, created_at=dt.datetime.now(dt.timezone.utc),
+        )
+        if persist:
+            self._op.save_run(run)
+            self._an.save_returns(run.run_id, prices.dates, r)
+        return run

src/fxhnt/cli.py

@@ -0,0 +1,69 @@
+"""CLI — the composition root. The ONLY place that wires concrete adapters to the application services.
+Swap an adapter here (Yahoo→Databento, SQLite→Postgres) without touching any other layer."""
+from __future__ import annotations
+
+import typer
+
+from fxhnt.adapters.data import YahooDataProvider
+from fxhnt.adapters.persistence import DuckDbAnalyticalStore, SqlOperationalRepository
+from fxhnt.application import ResearchService
+from fxhnt.config import Settings, get_settings
+from fxhnt.domain.models import AssetClass, Market, StrategySpec
+from fxhnt.domain.strategies import available
+
+app = typer.Typer(help="fxhnt — agentic strategy research & multi-strategy execution", no_args_is_help=True)
+
+
+def _build_service(settings: Settings) -> ResearchService:
+    return ResearchService(
+        data=YahooDataProvider(),
+        operational=SqlOperationalRepository(settings.operational_dsn),
+        analytical=DuckDbAnalyticalStore(settings.analytical_path),
+        settings=settings,
+    )
+
+
+@app.command()
+def research(
+    symbol: str = typer.Argument(..., help="ticker, e.g. SPY"),
+    kind: str = typer.Option("trend", help="strategy kind"),
+    asset_class: AssetClass = typer.Option(AssetClass.ETF),
+    window: int = typer.Option(200, help="trend MA window"),
+    n_trials: int = typer.Option(1, help="full search size (for the deflated-Sharpe correction)"),
+) -> None:
+    """Run one (market × strategy) candidate through the gauntlet and persist the verdict."""
+    settings = get_settings()
+    svc = _build_service(settings)
+    market = Market(symbol=symbol.upper(), asset_class=asset_class)
+    spec = StrategySpec(kind=kind, params={"window": float(window)})
+    run = svc.evaluate_candidate(market, spec, n_trials=n_trials)
+    s, v = run.stats, run.verdict
+    typer.echo(f"\n{market} | {spec.key()}")
+    typer.echo(f"  backtest: CAGR {100*s.cagr:+.1f}%  vol {100*s.ann_vol:.0f}%  Sharpe {s.sharpe:+.2f}  maxDD {100*s.max_drawdown:+.0f}%  (n={s.n_obs})")
+    typer.echo(f"  gauntlet: {v.summary()}")
+    for r in v.reasons:
+        typer.echo(f"     - {r}")
+    typer.echo(f"  -> {'PASS ✅' if v.passed else 'REJECT ❌'}  (run_id {run.run_id})\n")
+
+
+@app.command("list")
+def list_runs(passed_only: bool = typer.Option(False, "--passed-only")) -> None:
+    """List persisted research runs (the survivor library when --passed-only)."""
+    svc = _build_service(get_settings())
+    runs = svc._op.list_runs(passed_only=passed_only)  # noqa: SLF001 (composition root may reach in)
+    if not runs:
+        typer.echo("no runs yet.")
+        return
+    for run in runs:
+        flag = "PASS" if run.verdict.passed else "REJECT"
+        typer.echo(f"  [{flag}] {run.market} {run.spec.key()} | Sharpe {run.stats.sharpe:+.2f} DSR {run.verdict.dsr:.2f} | {run.run_id}")
+
+
+@app.command()
+def strategies() -> None:
+    """List available strategy kinds."""
+    typer.echo("available strategies: " + ", ".join(available()))
+
+
+if __name__ == "__main__":
+    app()

src/fxhnt/config.py

@@ -0,0 +1,47 @@
+"""Central configuration — env-driven, no secrets in code (pydantic-settings).
+
+Every layer reads its config from here; nothing hardcodes a connection string or threshold.
+Override via environment variables prefixed FXHNT_ (e.g. FXHNT_OPERATIONAL_DSN=postgresql+psycopg://...).
+"""
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+_DATA_DIR = Path.home() / ".fxhnt"
+
+
+class GauntletSettings(BaseSettings):
+    """Validation thresholds — the anti-overfitting bar. Conservative by design."""
+    model_config = SettingsConfigDict(env_prefix="FXHNT_GAUNTLET_")
+    dsr_min: float = 0.95            # deflated-Sharpe floor (accounts for the full search)
+    oos_min_sharpe: float = 0.0
+    max_is_oos_decay: float = 0.50   # OOS Sharpe must hold ≥ (1-this) × IS
+    oos_fraction: float = 0.40       # holdout fraction
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="FXHNT_", env_file=".env", extra="ignore")
+
+    # Operational store (relational: strategies, runs, verdicts, positions, trades).
+    # SQLite for dev/test; set to a postgresql+psycopg DSN in production.
+    operational_dsn: str = Field(default=f"sqlite:///{_DATA_DIR / 'operational.db'}")
+    # Analytical store (columnar: market data, backtest timeseries) — DuckDB embedded file.
+    analytical_path: str = Field(default=str(_DATA_DIR / "analytical.duckdb"))
+
+    cost_bps_per_turnover: float = 10.0   # round-trip cost model (bps of traded notional)
+    gauntlet: GauntletSettings = Field(default_factory=GauntletSettings)
+
+    def ensure_dirs(self) -> None:
+        _DATA_DIR.mkdir(parents=True, exist_ok=True)
+        Path(self.analytical_path).parent.mkdir(parents=True, exist_ok=True)
+
+
+@lru_cache
+def get_settings() -> Settings:
+    s = Settings()
+    s.ensure_dirs()
+    return s

src/fxhnt/domain/__init__.py

@@ -0,0 +1,3 @@
+"""Domain layer — pure business logic. NO I/O, no requests, no DB, no broker imports here.
+Everything in this package is deterministic and unit-testable in isolation.
+"""

src/fxhnt/domain/backtest.py

@@ -0,0 +1,37 @@
+"""Pure backtest engine — signal → net returns + stats. Deterministic, no I/O.
+
+Applies a round-trip cost model on turnover so a strategy that only works gross is exposed here.
+"""
+from __future__ import annotations
+
+import math
+
+import numpy as np
+
+from fxhnt.domain.models import BacktestResult, BacktestStats, PriceSeries, StrategySpec
+from fxhnt.domain.strategies.base import get_strategy
+
+
+def compute_stats(returns: np.ndarray, periods_per_year: int = 252) -> BacktestStats:
+    r = np.asarray(returns, float)
+    r = r[np.isfinite(r)]
+    if len(r) < 2:
+        return BacktestStats(cagr=0.0, ann_vol=0.0, sharpe=0.0, max_drawdown=0.0, n_obs=len(r))
+    ann, vol = float(r.mean() * periods_per_year), float(r.std() * math.sqrt(periods_per_year))
+    eq = np.cumprod(1.0 + r)
+    years = len(r) / periods_per_year
+    cagr = float(eq[-1] ** (1.0 / years) - 1.0) if years > 0 and eq[-1] > 0 else 0.0
+    dd = float((eq / np.maximum.accumulate(eq) - 1.0).min())
+    return BacktestStats(cagr=cagr, ann_vol=vol, sharpe=(ann / vol if vol > 0 else 0.0),
+                         max_drawdown=dd, n_obs=len(r))
+
+
+def run_backtest(prices: PriceSeries, spec: StrategySpec, cost_bps_per_turnover: float = 10.0) -> BacktestResult:
+    strategy = get_strategy(spec.kind)
+    pos = strategy.positions(prices, spec.params)
+    ret = prices.returns()
+    gross = pos * ret
+    turnover = np.abs(np.diff(np.concatenate([[0.0], pos])))
+    cost = turnover * (cost_bps_per_turnover / 1e4)
+    net = gross - cost
+    return BacktestResult(spec=spec, market=prices.market, returns=net, stats=compute_stats(net))

src/fxhnt/domain/gauntlet/__init__.py

@@ -0,0 +1,16 @@
+"""The validation gauntlet — fxhnt's anti-overfitting core (built and proven before any search)."""
+from fxhnt.domain.gauntlet.core import (
+    annualized_sharpe,
+    deflated_sharpe,
+    evaluate,
+    expected_max_sharpe,
+    probabilistic_sharpe,
+)
+
+__all__ = [
+    "annualized_sharpe",
+    "deflated_sharpe",
+    "evaluate",
+    "expected_max_sharpe",
+    "probabilistic_sharpe",
+]

src/fxhnt/domain/gauntlet/core.py

@@ -0,0 +1,101 @@
+"""Deflated-Sharpe validation math (Bailey & López de Prado 2014).
+
+The single most important guardrail in fxhnt: an agentic search over thousands of (strategy × market)
+combos is a false-positive factory unless the statistics correct for the full search size. The
+Deflated Sharpe Ratio answers "is this strategy real, GIVEN I tried N of them?". Pure stdlib + numpy.
+"""
+from __future__ import annotations
+
+import math
+from statistics import NormalDist
+
+import numpy as np
+
+from fxhnt.domain.models import Verdict
+
+_N = NormalDist()
+_EULER = 0.5772156649015329
+
+
+def _per_period_moments(r: np.ndarray) -> tuple[float, float, float, int]:
+    r = np.asarray(r, float)
+    r = r[np.isfinite(r)]
+    if len(r) < 3 or r.std(ddof=1) == 0:
+        return 0.0, 0.0, 3.0, len(r)
+    mu, sd = float(r.mean()), float(r.std(ddof=1))
+    z = (r - mu) / sd
+    return mu / sd, float((z ** 3).mean()), float((z ** 4).mean()), len(r)
+
+
+def annualized_sharpe(returns: np.ndarray, periods_per_year: int = 252) -> float:
+    r = np.asarray(returns, float)
+    r = r[np.isfinite(r)]
+    if len(r) < 2 or r.std() == 0:
+        return 0.0
+    return float(r.mean() / r.std() * math.sqrt(periods_per_year))
+
+
+def probabilistic_sharpe(returns: np.ndarray, sr_benchmark: float = 0.0) -> float:
+    """P(true per-period Sharpe > sr_benchmark), correcting for skew, kurtosis and sample length."""
+    sr, g1, g2, t = _per_period_moments(np.asarray(returns, float))
+    if t < 3:
+        return float("nan")
+    denom = math.sqrt(max(1e-12, 1.0 - g1 * sr + (g2 - 1.0) / 4.0 * sr * sr))
+    return _N.cdf((sr - sr_benchmark) * math.sqrt(t - 1) / denom)
+
+
+def expected_max_sharpe(n_trials: int, sr_variance: float) -> float:
+    """Expected MAX per-period Sharpe under n_trials independent strategies with the given cross-trial
+    Sharpe variance — the bar a 'winner' must clear to not be luck of the search."""
+    if n_trials < 2:
+        return 0.0
+    a = _N.inv_cdf(1.0 - 1.0 / n_trials)
+    b = _N.inv_cdf(1.0 - 1.0 / (n_trials * math.e))
+    return math.sqrt(max(sr_variance, 1e-12)) * ((1.0 - _EULER) * a + _EULER * b)
+
+
+def deflated_sharpe(returns: np.ndarray, n_trials: int, sr_variance: float) -> float:
+    """DSR ≈ P(this strategy is real | you searched n_trials). The anti-multiple-testing statistic."""
+    return probabilistic_sharpe(returns, expected_max_sharpe(n_trials, sr_variance))
+
+
+def evaluate(
+    is_returns: np.ndarray,
+    oos_returns: np.ndarray,
+    n_trials: int,
+    sr_variance: float,
+    *,
+    dsr_min: float = 0.95,
+    oos_min_sharpe: float = 0.0,
+    max_is_oos_decay: float = 0.50,
+    has_economic_rationale: bool | None = None,
+) -> Verdict:
+    """Full verdict. PASSES only if ALL hold:
+      1. DSR (over the FULL search) ≥ dsr_min       — not a multiple-testing artifact
+      2. OOS Sharpe > oos_min_sharpe                — works out of sample
+      3. OOS Sharpe ≥ (1 - max_is_oos_decay) × IS   — doesn't collapse OOS
+      4. (advisory) has_economic_rationale          — a structural reason the edge exists
+    """
+    dsr = deflated_sharpe(is_returns, n_trials, sr_variance)
+    is_sr, oos_sr = annualized_sharpe(is_returns), annualized_sharpe(oos_returns)
+    reasons: list[str] = []
+    ok = True
+    if not (dsr >= dsr_min):
+        ok = False
+        reasons.append(f"DSR {dsr:.3f} < {dsr_min} — likely a search artifact ({n_trials} trials)")
+    else:
+        reasons.append(f"DSR {dsr:.3f} ≥ {dsr_min} — survives multiple-testing correction")
+    if oos_sr <= oos_min_sharpe:
+        ok = False
+        reasons.append(f"OOS Sharpe {oos_sr:+.2f} ≤ {oos_min_sharpe} — no out-of-sample edge")
+    elif oos_sr < (1 - max_is_oos_decay) * is_sr:
+        ok = False
+        reasons.append(f"OOS Sharpe {oos_sr:+.2f} collapsed from IS {is_sr:+.2f}")
+    else:
+        reasons.append(f"OOS Sharpe {oos_sr:+.2f} holds vs IS {is_sr:+.2f}")
+    if has_economic_rationale is False:
+        ok = False
+        reasons.append("no structural rationale — refusing a purely statistical fit")
+    elif has_economic_rationale is None:
+        reasons.append("⚠ economic rationale UNVERIFIED — justify before deploying")
+    return Verdict(passed=ok, dsr=dsr, is_sharpe=is_sr, oos_sharpe=oos_sr, n_trials=n_trials, reasons=reasons)

src/fxhnt/domain/models.py

@@ -0,0 +1,112 @@
+"""Domain models — the contracts that flow between layers.
+
+Split by purpose:
+  * pydantic BaseModel for serializable contracts (DB / API / cross-boundary): Market, StrategySpec,
+    BacktestStats, Verdict, ResearchRun. Validated, immutable where it matters.
+  * frozen dataclass for in-memory numeric value objects holding numpy arrays (PriceSeries,
+    BacktestResult) — these never cross a serialization boundary as-is.
+"""
+from __future__ import annotations
+
+import datetime as dt
+from dataclasses import dataclass
+from enum import Enum
+
+import numpy as np
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class AssetClass(str, Enum):
+    EQUITY = "equity"
+    ETF = "etf"
+    FUTURE = "future"
+    FX = "fx"
+    CRYPTO = "crypto"
+    RATE = "rate"
+    COMMODITY = "commodity"
+
+
+class Market(BaseModel):
+    """A tradeable instrument identity."""
+    model_config = ConfigDict(frozen=True)
+    symbol: str
+    asset_class: AssetClass
+    venue: str = "SMART"
+    currency: str = "USD"
+
+    def __str__(self) -> str:
+        return f"{self.symbol}.{self.asset_class.value}"
+
+
+@dataclass(frozen=True)
+class PriceSeries:
+    """Adjusted-close series for one market, dates ascending. In-memory numeric value object."""
+    market: Market
+    dates: tuple[str, ...]
+    close: np.ndarray
+
+    def __post_init__(self) -> None:
+        if len(self.dates) != len(self.close):
+            raise ValueError("dates and close length mismatch")
+
+    def __len__(self) -> int:
+        return len(self.close)
+
+    def returns(self) -> np.ndarray:
+        r = np.zeros(len(self.close))
+        if len(self.close) > 1:
+            r[1:] = self.close[1:] / self.close[:-1] - 1.0
+        return r
+
+
+class StrategySpec(BaseModel):
+    """The recipe for a strategy: its kind + parameters. Hashable identity for the research matrix."""
+    model_config = ConfigDict(frozen=True)
+    kind: str
+    params: dict[str, float] = Field(default_factory=dict)
+
+    def key(self) -> str:
+        ps = ",".join(f"{k}={v:g}" for k, v in sorted(self.params.items()))
+        return f"{self.kind}({ps})"
+
+
+class BacktestStats(BaseModel):
+    cagr: float
+    ann_vol: float
+    sharpe: float
+    max_drawdown: float
+    n_obs: int
+
+
+@dataclass(frozen=True)
+class BacktestResult:
+    """Full backtest output — net daily returns + scalar stats."""
+    spec: StrategySpec
+    market: Market
+    returns: np.ndarray
+    stats: BacktestStats
+
+
+class Verdict(BaseModel):
+    """The gauntlet's ruling on a candidate, accounting for the full search size."""
+    passed: bool
+    dsr: float
+    is_sharpe: float
+    oos_sharpe: float
+    n_trials: int
+    reasons: list[str] = Field(default_factory=list)
+
+    def summary(self) -> str:
+        flag = "PASS" if self.passed else "REJECT"
+        return f"{flag} | DSR {self.dsr:.3f} | IS {self.is_sharpe:+.2f} | OOS {self.oos_sharpe:+.2f} | trials {self.n_trials}"
+
+
+class ResearchRun(BaseModel):
+    """A persisted record of one (market × strategy) evaluation through the pipeline."""
+    run_id: str
+    market: Market
+    spec: StrategySpec
+    stats: BacktestStats
+    verdict: Verdict
+    n_trials: int
+    created_at: dt.datetime

src/fxhnt/domain/strategies/__init__.py

@@ -0,0 +1,5 @@
+"""Strategy templates. Importing this package registers all built-in strategies."""
+from fxhnt.domain.strategies import trend  # noqa: F401  (import for side-effect: registration)
+from fxhnt.domain.strategies.base import Strategy, available, get_strategy, register
+
+__all__ = ["Strategy", "available", "get_strategy", "register"]

src/fxhnt/domain/strategies/base.py

@@ -0,0 +1,48 @@
+"""Strategy contract + registry. One Protocol, one registry — strategies plug in without the rest of
+the system knowing their internals (DRY: the research/backtest layers depend only on this contract)."""
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+import numpy as np
+
+from fxhnt.domain.models import PriceSeries
+
+
+@runtime_checkable
+class Strategy(Protocol):
+    kind: str
+    rationale: str  # the structural reason the edge should exist (the gauntlet requires one)
+
+    def positions(self, prices: PriceSeries, params: dict[str, float]) -> np.ndarray:
+        """Target position in [-1, 1] per day, set from info up to and including day t and applied to
+        day t+1's return (callers/implementations must avoid lookahead)."""
+        ...
+
+
+_REGISTRY: dict[str, Strategy] = {}
+
+
+def register(strategy: Strategy) -> Strategy:
+    _REGISTRY[strategy.kind] = strategy
+    return strategy
+
+
+def get_strategy(kind: str) -> Strategy:
+    if kind not in _REGISTRY:
+        raise KeyError(f"unknown strategy {kind!r}; available: {available()}")
+    return _REGISTRY[kind]
+
+
+def available() -> list[str]:
+    return sorted(_REGISTRY)
+
+
+def rolling_mean(x: np.ndarray, window: int) -> np.ndarray:
+    """Causal rolling mean (uses only x[..t]); shorter window at the start, no lookahead."""
+    cs = np.concatenate([[0.0], np.cumsum(x)])
+    out = np.empty(len(x))
+    for t in range(len(x)):
+        lo = max(0, t + 1 - window)
+        out[t] = (cs[t + 1] - cs[lo]) / (t + 1 - lo)
+    return out

src/fxhnt/domain/strategies/trend.py

@@ -0,0 +1,31 @@
+"""Time-series momentum (trend following) — a persistent, structurally-justified premium.
+
+Rationale: trend has a real economic basis (slow information diffusion + flow/herding) and is
+positively skewed crisis-alpha. Long when price is above its `window`-day moving average; flat
+(or short, if long_short) otherwise. Position set at close[t] applies to return[t+1] — no lookahead.
+"""
+from __future__ import annotations
+
+import numpy as np
+
+from fxhnt.domain.models import PriceSeries
+from fxhnt.domain.strategies.base import register, rolling_mean
+
+
+class TrendFollowing:
+    kind = "trend"
+    rationale = "time-series momentum: persistent premium from slow info diffusion + flow; crisis-alpha"
+
+    def positions(self, prices: PriceSeries, params: dict[str, float]) -> np.ndarray:
+        window = int(params.get("window", 200))
+        flat_value = -1.0 if params.get("long_short", 0.0) else 0.0
+        c = prices.close
+        ma = rolling_mean(c, window)
+        signal = np.where(c > ma, 1.0, flat_value)
+        signal[:window] = 0.0  # warmup: no position until the MA is meaningful
+        pos = np.zeros(len(c))
+        pos[1:] = signal[:-1]  # shift: decision at t-1 drives return at t (no lookahead)
+        return pos
+
+
+register(TrendFollowing())

src/fxhnt/ports/__init__.py

@@ -0,0 +1,8 @@
+"""Ports — abstract contracts (Protocols) between the domain/application and the outside world.
+Adapters implement these; the core depends only on the contracts. Swap Yahoo↔Databento or
+Postgres↔SQLite without touching domain logic.
+"""
+from fxhnt.ports.data import DataProvider
+from fxhnt.ports.repository import AnalyticalStore, OperationalRepository
+
+__all__ = ["DataProvider", "AnalyticalStore", "OperationalRepository"]

src/fxhnt/ports/data.py

@@ -0,0 +1,14 @@
+"""DataProvider contract — anything that can deliver an adjusted-close PriceSeries for a Market."""
+from __future__ import annotations
+
+from typing import Protocol
+
+from fxhnt.domain.models import Market, PriceSeries
+
+
+class DataProvider(Protocol):
+    name: str
+
+    def fetch(self, market: Market, start: str | None = None, end: str | None = None) -> PriceSeries:
+        """Return an ascending adjusted-close series. start/end are ISO dates (inclusive) or None."""
+        ...

src/fxhnt/ports/repository.py

@@ -0,0 +1,27 @@
+"""Persistence contracts. Two stores by purpose (per the chosen Postgres + DuckDB split):
+  * OperationalRepository — relational (Postgres/SQLite): research runs, verdicts, survivor library.
+  * AnalyticalStore       — columnar (DuckDB): market price series + backtest return timeseries.
+"""
+from __future__ import annotations
+
+from typing import Protocol
+
+import numpy as np
+
+from fxhnt.domain.models import Market, PriceSeries, ResearchRun
+
+
+class OperationalRepository(Protocol):
+    def save_run(self, run: ResearchRun) -> None: ...
+
+    def get_run(self, run_id: str) -> ResearchRun | None: ...
+
+    def list_runs(self, *, passed_only: bool = False) -> list[ResearchRun]: ...
+
+
+class AnalyticalStore(Protocol):
+    def save_prices(self, prices: PriceSeries) -> None: ...
+
+    def load_prices(self, market: Market) -> PriceSeries | None: ...
+
+    def save_returns(self, run_id: str, dates: tuple[str, ...], returns: np.ndarray) -> None: ...

tests/integration/test_research_slice.py

@@ -0,0 +1,55 @@
+"""End-to-end vertical slice with a FAKE data provider (no network) + temp SQLite + temp DuckDB.
+Proves data → backtest → gauntlet → persistence wires together and the domain stays infra-agnostic."""
+from __future__ import annotations
+
+import numpy as np
+
+from fxhnt.adapters.persistence import DuckDbAnalyticalStore, SqlOperationalRepository
+from fxhnt.application import ResearchService
+from fxhnt.config import GauntletSettings, Settings
+from fxhnt.domain.models import AssetClass, Market, PriceSeries, StrategySpec
+
+
+class FakeDataProvider:
+    name = "fake"
+
+    def __init__(self, prices: PriceSeries) -> None:
+        self._prices = prices
+
+    def fetch(self, market: Market, start: str | None = None, end: str | None = None) -> PriceSeries:
+        return self._prices
+
+
+def _trending_prices(market: Market, n: int = 2000, seed: int = 1) -> PriceSeries:
+    rng = np.random.default_rng(seed)
+    rets = rng.normal(0.0004, 0.01, n)  # genuine upward drift -> trend has something real to ride
+    close = 100.0 * np.cumprod(1.0 + rets)
+    dates = tuple(f"20{10 + i // 365:02d}-{1 + (i // 30) % 12:02d}-{1 + i % 28:02d}" for i in range(n))
+    return PriceSeries(market=market, dates=dates, close=close)
+
+
+def test_vertical_slice(tmp_path) -> None:
+    market = Market(symbol="TEST", asset_class=AssetClass.ETF)
+    settings = Settings(
+        operational_dsn=f"sqlite:///{tmp_path / 'op.db'}",
+        analytical_path=str(tmp_path / "an.duckdb"),
+        gauntlet=GauntletSettings(),
+    )
+    svc = ResearchService(
+        data=FakeDataProvider(_trending_prices(market)),
+        operational=SqlOperationalRepository(settings.operational_dsn),
+        analytical=DuckDbAnalyticalStore(settings.analytical_path),
+        settings=settings,
+    )
+
+    run = svc.evaluate_candidate(market, StrategySpec(kind="trend", params={"window": 100.0}), n_trials=1)
+
+    # the pipeline produced a coherent run
+    assert run.stats.n_obs > 1000
+    assert run.verdict.n_trials == 1
+    # it persisted to the operational store and round-trips
+    loaded = svc._op.get_run(run.run_id)  # noqa: SLF001
+    assert loaded is not None and loaded.run_id == run.run_id
+    assert loaded.verdict.passed == run.verdict.passed
+    # the analytical store cached the prices
+    assert svc._an.load_prices(market) is not None

tests/unit/test_gauntlet.py

@@ -0,0 +1,41 @@
+"""Falsification test for the gauntlet — the proof it's safe to scale a search on top of it.
+It MUST reject a best-of-N artifact mined on noise, and KEEP a genuine N=1 premium."""
+from __future__ import annotations
+
+import numpy as np
+
+from fxhnt.domain.gauntlet import annualized_sharpe, deflated_sharpe, evaluate
+
+RNG = np.random.default_rng(7)
+T = 2520
+N_TRIALS = 1000
+
+
+def test_rejects_best_of_n_on_noise() -> None:
+    """Mine the best of N random strategies on pure noise; the gauntlet must reject it."""
+    trials = RNG.normal(0.0, 0.01, (N_TRIALS, T))
+    srs = trials.mean(1) / trials.std(1)
+    best = int(np.argmax(srs))
+    best_r = trials[best]
+    sr_var = float(srs.var())
+    assert annualized_sharpe(best_r) > 0.8  # looks great in-sample by luck
+    split = int(0.6 * T)
+    v = evaluate(best_r[:split], best_r[split:], n_trials=N_TRIALS, sr_variance=sr_var, has_economic_rationale=False)
+    assert not v.passed, "gauntlet accepted an overfit best-of-N artifact"
+    assert v.dsr < 0.95
+
+
+def test_keeps_real_premium() -> None:
+    """A genuine positive-drift series, tested as one hypothesis, must pass."""
+    real = RNG.normal(0.0005, 0.01, T)
+    split = int(0.6 * T)
+    v = evaluate(real[:split], real[split:], n_trials=1, sr_variance=0.0, has_economic_rationale=True)
+    assert v.passed, "gauntlet rejected a genuine OOS-confirmed premium"
+    assert v.dsr >= 0.95
+
+
+def test_dsr_falls_as_trials_rise() -> None:
+    """The same returns get a strictly lower deflated Sharpe as the search size grows."""
+    r = RNG.normal(0.0004, 0.01, T)
+    sr_var = 0.04
+    assert deflated_sharpe(r, 1, sr_var) >= deflated_sharpe(r, 100, sr_var) >= deflated_sharpe(r, 10000, sr_var)