File size: 4,512 Bytes
ef20ebe | 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 116 117 118 | """
Covers the parts of data_layer.py that don't require network access:
validate_ohlcv (duplicate/invalid/NaN handling), LocalCache round-trip,
and data_freshness classification. fetch_ohlcv() itself needs a real
network call and is exercised only by the smoke-test script (expected
to fail here with DataSourceError -- see README).
"""
import numpy as np
import pandas as pd
from data_layer import LocalCache, data_freshness, validate_ohlcv
def test_validate_removes_duplicates_and_keeps_first():
idx = pd.to_datetime(["2024-01-01", "2024-01-01", "2024-01-02"], utc=True)
df = pd.DataFrame({
"open": [1, 999, 2], "high": [1, 999, 2], "low": [1, 999, 2],
"close": [1, 999, 2], "volume": [10, 10, 10],
}, index=idx)
out, report = validate_ohlcv(df)
assert report.duplicates_removed == 1
assert len(out) == 2
assert out.iloc[0]["close"] == 1 # first occurrence kept, not the duplicate
def test_validate_removes_invalid_ohlc_relationship():
idx = pd.date_range("2024-01-01", periods=3, freq="1h", tz="UTC")
df = pd.DataFrame({
"open": [10, 10, 10],
"high": [11, 9, 11], # row 1: high < open -> invalid
"low": [9, 9, 9],
"close": [10.5, 10.5, 10.5],
"volume": [5, 5, 5],
}, index=idx)
out, report = validate_ohlcv(df)
assert report.invalid_ohlc_removed == 1
assert len(out) == 2
def test_validate_removes_nan_rows_and_reports_them():
idx = pd.date_range("2024-01-01", periods=3, freq="1h", tz="UTC")
df = pd.DataFrame({
"open": [10, np.nan, 10], "high": [11, 11, 11], "low": [9, 9, 9],
"close": [10.5, 10.5, 10.5], "volume": [5, 5, 5],
}, index=idx)
out, report = validate_ohlcv(df)
assert report.nan_rows_removed == 1
assert len(out) == 2
def test_validate_marks_negative_volume_unavailable_not_invented():
idx = pd.date_range("2024-01-01", periods=2, freq="1h", tz="UTC")
df = pd.DataFrame({
"open": [10, 10], "high": [11, 11], "low": [9, 9],
"close": [10.5, 10.5], "volume": [5, -3],
}, index=idx)
out, _ = validate_ohlcv(df)
assert np.isnan(out.iloc[1]["volume"]), "negative volume must become NaN, never a guessed positive number"
def test_local_cache_round_trip(tmp_path_str="/tmp/_moirai_test_cache.sqlite3"):
import os
if os.path.exists(tmp_path_str):
os.remove(tmp_path_str)
cache = LocalCache(tmp_path_str)
idx = pd.date_range("2024-01-01", periods=3, freq="1h", tz="UTC")
df = pd.DataFrame({"open": [1, 2, 3], "high": [1, 2, 3], "low": [1, 2, 3],
"close": [1, 2, 3], "volume": [1, 2, 3]}, index=idx)
assert cache.get("EURUSD=X", "1h", "yfinance", "2024-01-01", "2024-01-02") is None
cache.set("EURUSD=X", "1h", "yfinance", "2024-01-01", "2024-01-02", df)
round_tripped = cache.get("EURUSD=X", "1h", "yfinance", "2024-01-01", "2024-01-02")
assert round_tripped is not None
assert len(round_tripped) == 3
assert list(round_tripped["close"]) == [1, 2, 3]
os.remove(tmp_path_str)
def test_data_freshness_classification():
now = pd.Timestamp.now(tz="UTC")
fresh = data_freshness(now - pd.Timedelta(minutes=1), "1h")
stale = data_freshness(now - pd.Timedelta(hours=10), "1h")
assert fresh["status"] == "fresh"
assert stale["status"] == "stale"
def test_resolve_history_window_days():
from data_layer import resolve_history_window
kind, value = resolve_history_window("1 day")
assert kind == "start"
kind, value = resolve_history_window("30 days")
assert kind == "start"
def test_resolve_history_window_months_uses_calendar_months():
from data_layer import resolve_history_window
from datetime import datetime, timezone
kind, value = resolve_history_window("3 months")
assert kind == "start"
now = datetime.now(timezone.utc)
delta_days = (now - value).days
# 3 calendar months is 89-92 days depending on which months are spanned --
# NOT exactly 90 (which a naive 30*3 approximation would assume).
assert 88 <= delta_days <= 93, f"expected ~3 calendar months, got {delta_days} days"
def test_resolve_history_window_max():
from data_layer import resolve_history_window
kind, value = resolve_history_window("max")
assert kind == "period"
assert value == "max"
def test_resolve_history_window_years():
from data_layer import resolve_history_window
kind, value = resolve_history_window("2 years")
assert kind == "start"
|