0ca2e5284a
application/agent/hunter.AgentHunter: an LLM proposes the next batch of (market, strategy, window)
candidates, reasoning about strategy-market fit and the gaps the current population leaves — clamped to
the real universe + known strategies so hallucinations can't enter. FactoryRuntime now takes a
territory_provider (callable): fixed grids OR the agent's per-cycle proposals. CLI run --hunters agent.
Verified on local qwen2.5:3b: proposed 3 FOCUSED candidates (vs 18 grid) -> global trials 3 not 18 (the
discipline win of agentic search: lower multiple-testing burden) -> found buy_hold on GC in
trend_down|high_vol (gold flight-to-safety in panic). fxhnt 1+2 complete: [hunt] 6 new candidates → 1 specialists | global hypotheses 9 | by regime {'trend_down|high_vol': 1}
[book] as of 2026-06-08 | 0 active, 2 dormant | weights (flat)
on cron = the standing agentic force that constantly seeks regime-specialists and runs them side by side.
22/22 tests.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
38 lines
1.9 KiB
Python
38 lines
1.9 KiB
Python
"""Factory runtime: one cycle hunts then assembles; re-running dedups (honest global trial count)."""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
|
|
from fxhnt.adapters.persistence import DuckDbAnalyticalStore, SqlStrategyStore
|
|
from fxhnt.application import FactoryRuntime, Territory
|
|
from fxhnt.config import Settings
|
|
from fxhnt.domain.factory import StrategyStatus
|
|
from fxhnt.domain.models import AssetClass, Market, PriceSeries
|
|
|
|
|
|
class FakeData:
|
|
name = "fake"
|
|
|
|
def fetch(self, market: Market, start=None, end=None) -> PriceSeries:
|
|
rng = np.random.default_rng(sum(ord(c) for c in market.symbol))
|
|
close = 100.0 * np.cumprod(1.0 + rng.normal(0.001, 0.01, 1600))
|
|
return PriceSeries(market=market, dates=tuple(str(i) for i in range(1600)), close=close)
|
|
|
|
|
|
def test_cycle_hunts_assembles_and_dedups(tmp_path) -> None:
|
|
settings = Settings(analytical_path=str(tmp_path / "an.duckdb"), operational_dsn=f"sqlite:///{tmp_path / 'r.db'}")
|
|
store = SqlStrategyStore(settings.operational_dsn)
|
|
territories = [Territory(name="trend-hunter", kind="trend", markets=["AAA", "BBB"],
|
|
asset_class=AssetClass.FUTURE, param_grid=[{"window": 100.0}])]
|
|
rt = FactoryRuntime(FakeData(), DuckDbAnalyticalStore(settings.analytical_path), store, settings,
|
|
territory_provider=lambda: territories, book_status=StrategyStatus.DISCOVERED)
|
|
|
|
c1 = rt.cycle()
|
|
assert c1.fleet.candidates_tested == 2 and c1.fleet.global_trials == 2
|
|
assert c1.fleet.specialists_found >= 1
|
|
assert isinstance(c1.book.target_weights, dict) # a book was assembled
|
|
|
|
c2 = rt.cycle() # re-run: everything already seen
|
|
assert c2.fleet.candidates_tested == 0 # dedup — no re-testing
|
|
assert c2.fleet.global_trials == 2 # global count stays honest
|