File size: 4,042 Bytes
4d68493 6b66ac0 4d68493 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """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
# ---------------------------------------------------------------------------
# Synthetic OHLCV helpers
# ---------------------------------------------------------------------------
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()
|