BuckLakeAI / tests /test_raw_predict_contract.py
Parker's Fedora
Infer default model version from R45 artifact name.
9a1462b
Raw
History Blame Contribute Delete
4.48 kB
import tempfile
import unittest
from pathlib import Path
from bucklake_ai.config import get_settings
from bucklake_ai.schemas import PredictRequest
def _bar(offset: float) -> dict:
return {
"open": 100.0 + offset,
"close": 101.0 + offset,
"high": 102.0 + offset,
"low": 99.0 + offset,
"volume": 1000.0 + offset,
"amount": 5000.0 + offset,
}
def _bars() -> list[dict]:
return [_bar(float(i)) for i in range(30)]
class PredictRequestSchemaTests(unittest.TestCase):
def test_accepts_raw_input_contract(self):
request = PredictRequest.model_validate(
{
"request_id": "req-1",
"symbol": "aapl",
"published_at": "2026-04-15T12:30:00Z",
"text_session": "post_market",
"market_bars": {
"stock": _bars(),
"inx": _bars(),
"dj": _bars(),
"ixic": _bars(),
"ndx": _bars(),
},
"history_news": [
{
"text": "Older Apple news",
"published_at": "2026-04-14T10:00:00Z",
"symbols": ["AAPL", "MSFT"],
}
],
"options": {
"include_raw_heads": True,
"include_debug_meta": False,
},
}
)
self.assertEqual(request.symbol, "AAPL")
self.assertEqual(request.text, "")
self.assertEqual(request.text_session, "post_market")
self.assertEqual(len(request.market_bars.stock), 30)
self.assertEqual(request.history_news[0].symbols, ["AAPL", "MSFT"])
def test_accepts_empty_text(self):
request = PredictRequest.model_validate(
{
"symbol": "AAPL",
"published_at": "2026-04-15T12:30:00Z",
"text": " ",
"market_bars": {
"stock": _bars(),
"inx": _bars(),
"dj": _bars(),
"ixic": _bars(),
"ndx": _bars(),
},
}
)
self.assertEqual(request.text, "")
def test_rejects_history_news_without_symbols(self):
with self.assertRaises(Exception):
PredictRequest.model_validate(
{
"symbol": "AAPL",
"published_at": "2026-04-15T12:30:00Z",
"text": "Apple launches something new.",
"market_bars": {
"stock": _bars(),
"inx": _bars(),
"dj": _bars(),
"ixic": _bars(),
"ndx": _bars(),
},
"history_news": [
{
"text": "Older Apple news",
"published_at": "2026-04-14T10:00:00Z",
"symbols": [],
}
],
}
)
class DefaultSettingsTests(unittest.TestCase):
def test_defaults_point_to_r45_best_anchored_path_model(self):
get_settings.cache_clear()
settings = get_settings()
self.assertEqual(settings.model_version, "anchored-path-v2")
self.assertEqual(
settings.hf_model_filename,
"stock_prediction_model_anchored-path-v2_best_round45.keras",
)
self.assertEqual(
settings.model_weights_path.name,
"stock_prediction_model_anchored-path-v2_best_round45.keras",
)
def test_defaults_point_to_round41_scaler_artifact(self):
get_settings.cache_clear()
settings = get_settings()
self.assertEqual(settings.hf_scaler_filename, "scalers_v21_rolling_7d.json")
self.assertEqual(
settings.scaler_artifact_path.name, "scalers_v21_rolling_7d.json"
)
self.assertEqual(settings.scaler_artifact_path.parent.name, "data")
def test_finbert_sentiment_is_enabled_by_default(self):
get_settings.cache_clear()
settings = get_settings()
self.assertTrue(settings.enable_finbert_sentiment)
self.assertEqual(settings.finbert_model_name, "ProsusAI/finbert")
if __name__ == "__main__":
unittest.main()