| """Shared pytest fixtures. |
| |
| Every test gets an isolated ``tmp_path`` and we monkeypatch every path |
| attribute in :mod:`scanner.paths` to point inside that directory. Since |
| every consuming module reads paths via ``paths.X`` (rather than |
| ``from .paths import X``) this is sufficient to isolate disk I/O per |
| test - no module reload tricks required. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import os |
| from datetime import datetime, timedelta |
|
|
| import numpy as np |
| import pandas as pd |
| import pytest |
|
|
| from scanner import paths |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def isolated_paths(tmp_path, monkeypatch): |
| """Redirect every persistent path under :mod:`scanner.paths` into ``tmp_path``.""" |
| history_dir = tmp_path / "history" |
| history_dir.mkdir() |
| monkeypatch.setattr(paths, "TMP_DIR", str(tmp_path)) |
| monkeypatch.setattr(paths, "HISTORY_DIR", str(history_dir)) |
| monkeypatch.setattr(paths, "CACHE_PATH", str(tmp_path / "ohlcv.parquet")) |
| monkeypatch.setattr(paths, "SECTOR_CACHE_PATH", |
| str(tmp_path / "sectors.parquet")) |
| monkeypatch.setattr(paths, "WATCHLIST_PATH", |
| str(tmp_path / "watchlist.json")) |
| monkeypatch.setattr(paths, "LEARNED_WEIGHTS_PATH", |
| str(tmp_path / "learned.json")) |
| monkeypatch.setattr(paths, "PERFORMANCE_LOG_PATH", |
| str(tmp_path / "perf.parquet")) |
| monkeypatch.setattr(paths, "RESULTS_CSV_PATH", |
| str(tmp_path / "results.csv")) |
| monkeypatch.setattr(paths, "L2_CACHE_PATH", |
| str(tmp_path / "l2.parquet")) |
| monkeypatch.setattr(paths, "OPTIONS_CACHE_PATH", |
| str(tmp_path / "options.parquet")) |
| monkeypatch.setattr(paths, "TICK_CACHE_DIR", |
| str(tmp_path / "ticks")) |
| monkeypatch.setattr(paths, "INTRADAY_CACHE_DIR", |
| str(tmp_path / "intraday")) |
| monkeypatch.setattr(paths, "STUB_DIR", |
| str(tmp_path / "stubs")) |
| yield |
|
|
|
|
| |
| |
| |
|
|
| def _business_days(n: int, end: datetime | None = None) -> list[datetime]: |
| end = end or datetime(2026, 1, 30) |
| out = [] |
| d = end |
| while len(out) < n: |
| if d.weekday() < 5: |
| out.append(d) |
| d -= timedelta(days=1) |
| return list(reversed(out)) |
|
|
|
|
| def make_uptrend(n: int = 120, start_price: float = 50.0, |
| daily_drift: float = 0.003, |
| vol_base: int = 1_000_000, |
| seed: int = 1) -> pd.DataFrame: |
| """Generate a synthetic OHLCV frame with a clear up-trend and increasing |
| on-balance volume (close > prev_close most days).""" |
| rng = np.random.default_rng(seed) |
| dates = _business_days(n) |
| close = [start_price] |
| for _ in range(1, n): |
| ret = daily_drift + rng.normal(0, 0.008) |
| close.append(close[-1] * (1.0 + ret)) |
| close = np.array(close) |
| open_ = close * (1 + rng.normal(0, 0.002, size=n)) |
| high = np.maximum(open_, close) * (1 + np.abs(rng.normal(0, 0.005, size=n))) |
| low = np.minimum(open_, close) * (1 - np.abs(rng.normal(0, 0.005, size=n))) |
| vol = (vol_base * (1 + rng.normal(0, 0.2, size=n))).clip(1e4).astype(int) |
| df = pd.DataFrame({"Date": dates, "Open": open_, "High": high, |
| "Low": low, "Close": close, "Volume": vol}) |
| return df |
|
|
|
|
| def make_downtrend(**kwargs) -> pd.DataFrame: |
| kwargs.setdefault("daily_drift", -0.003) |
| kwargs.setdefault("seed", 2) |
| return make_uptrend(**kwargs) |
|
|
|
|
| def make_flat(**kwargs) -> pd.DataFrame: |
| kwargs.setdefault("daily_drift", 0.0) |
| kwargs.setdefault("seed", 3) |
| return make_uptrend(**kwargs) |
|
|
|
|
| @pytest.fixture |
| def uptrend_frame() -> pd.DataFrame: |
| return make_uptrend() |
|
|
|
|
| @pytest.fixture |
| def downtrend_frame() -> pd.DataFrame: |
| return make_downtrend() |
|
|
|
|
| @pytest.fixture |
| def flat_frame() -> pd.DataFrame: |
| return make_flat() |
|
|
