Sync from GitHub

tests/test_ai_engine.py

@@ -0,0 +1,267 @@
+"""
+Tests for AI Engine components.
+"""
+
+import pytest
+import numpy as np
+import pandas as pd
+from datetime import datetime, timezone, timedelta
+from unittest.mock import patch, MagicMock
+
+
+class TestFinBERTScoring:
+    """Tests for FinBERT sentiment scoring."""
+    
+    def test_score_text_empty_input(self):
+        """Test scoring with empty input."""
+        from app.ai_engine import score_text_with_finbert
+        
+        # Mock pipeline
+        mock_pipe = MagicMock()
+        
+        # Empty text should return neutral scores
+        result = score_text_with_finbert(mock_pipe, "")
+        
+        assert result["prob_positive"] == 0.33
+        assert result["prob_neutral"] == 0.34
+        assert result["prob_negative"] == 0.33
+        assert result["score"] == 0.0
+    
+    def test_score_text_short_input(self):
+        """Test scoring with very short input."""
+        from app.ai_engine import score_text_with_finbert
+        
+        mock_pipe = MagicMock()
+        
+        # Short text (< 10 chars) should return neutral
+        result = score_text_with_finbert(mock_pipe, "hi")
+        
+        assert result["score"] == 0.0
+    
+    def test_score_text_normal_input(self):
+        """Test scoring with normal input."""
+        from app.ai_engine import score_text_with_finbert
+        
+        # Mock pipeline to return positive sentiment
+        mock_pipe = MagicMock()
+        mock_pipe.return_value = [[
+            {"label": "positive", "score": 0.8},
+            {"label": "neutral", "score": 0.15},
+            {"label": "negative", "score": 0.05},
+        ]]
+        
+        result = score_text_with_finbert(
+            mock_pipe, 
+            "Copper prices surge to new highs on strong demand"
+        )
+        
+        assert result["prob_positive"] == 0.8
+        assert result["prob_neutral"] == 0.15
+        assert result["prob_negative"] == 0.05
+        assert result["score"] == 0.75  # 0.8 - 0.05
+    
+    def test_score_text_negative_sentiment(self):
+        """Test scoring with negative sentiment."""
+        from app.ai_engine import score_text_with_finbert
+        
+        mock_pipe = MagicMock()
+        mock_pipe.return_value = [[
+            {"label": "positive", "score": 0.1},
+            {"label": "neutral", "score": 0.2},
+            {"label": "negative", "score": 0.7},
+        ]]
+        
+        result = score_text_with_finbert(
+            mock_pipe,
+            "Copper prices crash amid recession fears"
+        )
+        
+        assert result["score"] == -0.6  # 0.1 - 0.7
+
+
+class TestSentimentAggregation:
+    """Tests for sentiment aggregation logic."""
+    
+    def test_recency_weighting(self):
+        """Test that later articles get higher weight."""
+        # This tests the concept, actual implementation may vary
+        tau = 12.0
+        
+        # Article at 9am vs 4pm
+        hours_early = 9.0
+        hours_late = 16.0
+        
+        weight_early = np.exp(hours_early / tau)
+        weight_late = np.exp(hours_late / tau)
+        
+        # Later article should have higher weight
+        assert weight_late > weight_early
+    
+    def test_weighted_average_calculation(self):
+        """Test weighted average calculation."""
+        scores = np.array([0.5, -0.2, 0.3])
+        weights = np.array([0.2, 0.3, 0.5])  # Normalized weights
+        
+        weighted_avg = np.sum(scores * weights)
+        expected = 0.5 * 0.2 + (-0.2) * 0.3 + 0.3 * 0.5
+        
+        assert abs(weighted_avg - expected) < 1e-10
+    
+    def test_sentiment_index_range(self):
+        """Test that sentiment index is in valid range."""
+        # Sentiment index should be between -1 and 1
+        scores = np.array([0.9, -0.8, 0.5])
+        weights = np.array([0.33, 0.33, 0.34])
+        
+        weighted_avg = np.sum(scores * weights)
+        
+        assert -1 <= weighted_avg <= 1
+
+
+class TestFeatureEngineering:
+    """Tests for feature engineering."""
+    
+    def test_technical_indicators(self, sample_price_data):
+        """Test that technical indicators are calculated correctly."""
+        df = sample_price_data
+        
+        # Calculate SMA
+        sma_5 = df["close"].rolling(window=5).mean()
+        sma_10 = df["close"].rolling(window=10).mean()
+        
+        # SMA calculations should not be NaN after sufficient data
+        assert not np.isnan(sma_5.iloc[-1])
+        assert not np.isnan(sma_10.iloc[-1])
+        
+        # SMA10 should smooth more than SMA5
+        assert sma_10.std() < df["close"].std()
+    
+    def test_return_calculation(self, sample_price_data):
+        """Test return calculation."""
+        df = sample_price_data
+        
+        # Calculate returns
+        returns = df["close"].pct_change()
+        
+        # First return should be NaN
+        assert np.isnan(returns.iloc[0])
+        
+        # Returns should be small (reasonable daily returns)
+        assert abs(returns.iloc[1:].mean()) < 0.1
+    
+    def test_volatility_calculation(self, sample_price_data):
+        """Test volatility calculation."""
+        df = sample_price_data
+        
+        returns = df["close"].pct_change()
+        volatility_10 = returns.rolling(window=10).std()
+        
+        # Volatility should be positive
+        assert all(v >= 0 or np.isnan(v) for v in volatility_10)
+    
+    def test_lagged_features(self, sample_price_data):
+        """Test lagged feature creation."""
+        df = sample_price_data
+        
+        returns = df["close"].pct_change()
+        
+        # Create lags
+        lag_1 = returns.shift(1)
+        lag_2 = returns.shift(2)
+        lag_3 = returns.shift(3)
+        
+        # Lags should have correct offset
+        assert lag_1.iloc[5] == returns.iloc[4]
+        assert lag_2.iloc[5] == returns.iloc[3]
+        assert lag_3.iloc[5] == returns.iloc[2]
+
+
+class TestModelTraining:
+    """Tests for model training logic."""
+    
+    def test_train_test_split_temporal(self):
+        """Test that train/test split respects time order."""
+        dates = pd.date_range(start="2025-01-01", periods=100, freq="D")
+        
+        validation_days = 20
+        split_date = dates.max() - timedelta(days=validation_days)
+        
+        train_dates = dates[dates <= split_date]
+        val_dates = dates[dates > split_date]
+        
+        # All train dates should be before all val dates
+        assert train_dates.max() < val_dates.min()
+        
+        # Correct number of validation samples
+        assert len(val_dates) == validation_days
+    
+    def test_feature_importance_normalized(self):
+        """Test that feature importance sums to 1."""
+        importance = {
+            "feature_a": 10.0,
+            "feature_b": 5.0,
+            "feature_c": 3.0,
+            "feature_d": 2.0,
+        }
+        
+        total = sum(importance.values())
+        normalized = {k: v / total for k, v in importance.items()}
+        
+        assert abs(sum(normalized.values()) - 1.0) < 1e-10
+    
+    def test_prediction_direction_from_return(self):
+        """Test prediction direction logic."""
+        def get_direction(predicted_return, threshold=0.005):
+            if predicted_return > threshold:
+                return "up"
+            elif predicted_return < -threshold:
+                return "down"
+            else:
+                return "neutral"
+        
+        assert get_direction(0.02) == "up"
+        assert get_direction(-0.02) == "down"
+        assert get_direction(0.001) == "neutral"
+        assert get_direction(-0.003) == "neutral"
+
+
+class TestModelPersistence:
+    """Tests for model saving and loading."""
+    
+    def test_model_path_generation(self):
+        """Test model path generation."""
+        from datetime import datetime
+        
+        target_symbol = "HG=F"
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        
+        model_filename = f"xgb_{target_symbol.replace('=', '_')}_{timestamp}.json"
+        latest_filename = f"xgb_{target_symbol.replace('=', '_')}_latest.json"
+        
+        assert "HG_F" in model_filename
+        assert "HG_F" in latest_filename
+        assert model_filename.endswith(".json")
+    
+    def test_metrics_json_structure(self):
+        """Test that metrics JSON has required fields."""
+        import json
+        
+        metrics = {
+            "target_symbol": "HG=F",
+            "trained_at": datetime.now(timezone.utc).isoformat(),
+            "train_samples": 200,
+            "val_samples": 30,
+            "train_mae": 0.01,
+            "train_rmse": 0.015,
+            "val_mae": 0.02,
+            "val_rmse": 0.025,
+            "best_iteration": 50,
+            "feature_count": 58,
+        }
+        
+        # Should serialize properly
+        json_str = json.dumps(metrics)
+        loaded = json.loads(json_str)
+        
+        assert loaded["target_symbol"] == "HG=F"
+        assert loaded["val_mae"] == 0.02

tests/test_api.py

@@ -0,0 +1,250 @@
+"""
+Tests for API endpoints.
+"""
+
+import pytest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timezone
+
+
+class TestHealthEndpoint:
+    """Tests for /api/health endpoint."""
+    
+    def test_health_response_structure(self):
+        """Test that health response has required fields."""
+        from app.schemas import HealthResponse
+        
+        response = HealthResponse(
+            status="healthy",
+            db_type="postgresql",
+            models_found=1,
+            pipeline_locked=False,
+            timestamp=datetime.now(timezone.utc).isoformat(),
+            news_count=100,
+            price_bars_count=500
+        )
+        
+        assert response.status == "healthy"
+        assert response.db_type == "postgresql"
+        assert response.models_found == 1
+        assert response.pipeline_locked is False
+        assert response.news_count == 100
+        assert response.price_bars_count == 500
+    
+    def test_health_status_degraded_no_models(self):
+        """Test degraded status when no models found."""
+        from app.schemas import HealthResponse
+        
+        response = HealthResponse(
+            status="degraded",
+            db_type="postgresql",
+            models_found=0,
+            pipeline_locked=False,
+            timestamp=datetime.now(timezone.utc).isoformat(),
+        )
+        
+        assert response.status == "degraded"
+        assert response.models_found == 0
+
+
+class TestAnalysisSchema:
+    """Tests for analysis report schema."""
+    
+    def test_analysis_report_structure(self):
+        """Test AnalysisReport schema validation."""
+        from app.schemas import AnalysisReport, Influencer
+        
+        influencers = [
+            Influencer(feature="HG=F_EMA_10", importance=0.15, description="Test"),
+            Influencer(feature="DX-Y.NYB_ret1", importance=0.10, description="Test"),
+        ]
+        
+        report = AnalysisReport(
+            symbol="HG=F",
+            prediction_direction="up",
+            confidence_score=0.75,
+            current_price=4.25,
+            predicted_return=0.015,
+            sentiment_index=0.35,
+            news_count_24h=15,
+            model_metrics={
+                "val_mae": 0.02,
+                "val_rmse": 0.025,
+            },
+            top_influencers=influencers,
+            generated_at=datetime.now(timezone.utc).isoformat(),
+        )
+        
+        assert report.symbol == "HG=F"
+        assert report.prediction_direction == "up"
+        assert report.confidence_score == 0.75
+        assert len(report.top_influencers) == 2
+    
+    def test_prediction_direction_values(self):
+        """Test valid prediction directions."""
+        from app.schemas import AnalysisReport
+        
+        for direction in ["up", "down", "neutral"]:
+            report = AnalysisReport(
+                symbol="HG=F",
+                prediction_direction=direction,
+                confidence_score=0.5,
+                current_price=4.0,
+                predicted_return=0.0,
+                sentiment_index=0.0,
+                news_count_24h=0,
+                model_metrics={},
+                top_influencers=[],
+                generated_at=datetime.now(timezone.utc).isoformat(),
+            )
+            assert report.prediction_direction == direction
+
+
+class TestHistorySchema:
+    """Tests for history response schema."""
+    
+    def test_history_data_point(self):
+        """Test HistoryDataPoint schema."""
+        from app.schemas import HistoryDataPoint
+        
+        point = HistoryDataPoint(
+            date="2026-01-01",
+            price=4.25,
+            sentiment_index=0.35,
+            sentiment_news_count=10,
+        )
+        
+        assert point.date == "2026-01-01"
+        assert point.price == 4.25
+        assert point.sentiment_index == 0.35
+        assert point.sentiment_news_count == 10
+    
+    def test_history_data_point_nullable_sentiment(self):
+        """Test that sentiment can be None."""
+        from app.schemas import HistoryDataPoint
+        
+        point = HistoryDataPoint(
+            date="2026-01-01",
+            price=4.25,
+            sentiment_index=None,
+            sentiment_news_count=None,
+        )
+        
+        assert point.sentiment_index is None
+        assert point.sentiment_news_count is None
+    
+    def test_history_response(self):
+        """Test HistoryResponse schema."""
+        from app.schemas import HistoryResponse, HistoryDataPoint
+        
+        data = [
+            HistoryDataPoint(date="2026-01-01", price=4.20),
+            HistoryDataPoint(date="2026-01-02", price=4.25),
+        ]
+        
+        response = HistoryResponse(symbol="HG=F", data=data)
+        
+        assert response.symbol == "HG=F"
+        assert len(response.data) == 2
+
+
+class TestPipelineLock:
+    """Tests for pipeline lock mechanism."""
+    
+    def test_lock_acquire_release(self, tmp_path):
+        """Test acquiring and releasing lock."""
+        from app.lock import PipelineLock
+        
+        lock_file = tmp_path / "test.lock"
+        lock = PipelineLock(lock_file=str(lock_file), timeout=0)
+        
+        # Should acquire
+        assert lock.acquire() is True
+        assert lock_file.exists()
+        
+        # Should release
+        lock.release()
+        assert not lock_file.exists()
+    
+    def test_lock_already_held(self, tmp_path):
+        """Test that second acquire fails when lock is held."""
+        from app.lock import PipelineLock
+        
+        lock_file = tmp_path / "test.lock"
+        lock1 = PipelineLock(lock_file=str(lock_file), timeout=0)
+        lock2 = PipelineLock(lock_file=str(lock_file), timeout=0)
+        
+        # First lock should succeed
+        assert lock1.acquire() is True
+        
+        # Second lock should fail
+        assert lock2.acquire() is False
+        
+        # Cleanup
+        lock1.release()
+    
+    def test_is_pipeline_locked(self, tmp_path):
+        """Test is_pipeline_locked helper."""
+        from app.lock import PipelineLock
+        
+        lock_file = tmp_path / "test.lock"
+        
+        with patch("app.lock.get_settings") as mock_settings:
+            mock_settings.return_value.pipeline_lock_file = str(lock_file)
+            
+            from app.lock import is_pipeline_locked
+            
+            # Initially not locked
+            assert is_pipeline_locked() is False
+            
+            # Create lock
+            lock_file.write_text("locked")
+            assert is_pipeline_locked() is True
+            
+            # Remove lock
+            lock_file.unlink()
+            assert is_pipeline_locked() is False
+
+
+class TestDataNormalization:
+    """Tests for URL and text normalization."""
+    
+    def test_normalize_url(self):
+        """Test URL normalization."""
+        from app.utils import normalize_url
+        
+        # Should remove tracking params
+        url = "https://example.com/article?id=123&utm_source=google&utm_medium=cpc"
+        normalized = normalize_url(url)
+        
+        assert "utm_source" not in normalized
+        assert "utm_medium" not in normalized
+        assert "id=123" in normalized
+    
+    def test_generate_dedup_key(self):
+        """Test dedup key generation."""
+        from app.utils import generate_dedup_key
+        
+        key1 = generate_dedup_key("Copper prices rise", "https://example.com/a")
+        key2 = generate_dedup_key("Copper prices rise", "https://example.com/a")
+        key3 = generate_dedup_key("Different title", "https://example.com/a")
+        
+        # Same input should give same key
+        assert key1 == key2
+        
+        # Different input should give different key
+        assert key1 != key3
+    
+    def test_truncate_text(self):
+        """Test text truncation."""
+        from app.utils import truncate_text
+        
+        long_text = "a" * 1000
+        truncated = truncate_text(long_text, max_length=100)
+        
+        assert len(truncated) == 100
+        
+        short_text = "hello"
+        not_truncated = truncate_text(short_text, max_length=100)
+        
+        assert not_truncated == "hello"

tests/test_data_ingestion.py

@@ -0,0 +1,233 @@
+"""
+Tests for data ingestion and management.
+"""
+
+import pytest
+from datetime import datetime, timezone, timedelta
+from unittest.mock import patch, MagicMock
+
+
+class TestLanguageDetection:
+    """Tests for language detection."""
+    
+    def test_detect_english(self):
+        """Test detection of English text."""
+        from app.data_manager import detect_language
+        
+        result = detect_language("Copper prices rose sharply today")
+        assert result == "en"
+    
+    def test_detect_non_english(self):
+        """Test detection of non-English text."""
+        from app.data_manager import detect_language
+        
+        # German
+        result = detect_language("Die Kupferpreise sind heute gestiegen")
+        assert result != "en"
+    
+    def test_detect_empty_text(self):
+        """Test detection with empty text."""
+        from app.data_manager import detect_language
+        
+        result = detect_language("")
+        assert result is None
+    
+    def test_detect_short_text(self):
+        """Test detection with very short text."""
+        from app.data_manager import detect_language
+        
+        # Short text may fail detection
+        result = detect_language("Hi")
+        # Should handle gracefully
+        assert result is None or isinstance(result, str)
+
+
+class TestLanguageFiltering:
+    """Tests for language filtering."""
+    
+    def test_filter_keeps_english(self, sample_articles):
+        """Test that English articles are kept."""
+        from app.data_manager import filter_by_language
+        
+        articles = [
+            {"title": "Copper prices rise", "description": "Copper up today"},
+            {"title": "Mining output increases", "description": "Good news"},
+        ]
+        
+        filtered, count = filter_by_language(articles, "en")
+        
+        assert len(filtered) == 2
+        assert count == 0
+    
+    def test_filter_removes_non_english(self):
+        """Test that non-English articles are filtered."""
+        from app.data_manager import filter_by_language
+        
+        articles = [
+            {"title": "Copper prices rise", "description": "Copper up today"},
+            {"title": "Kupferpreise steigen", "description": "Kupfer heute höher"},
+        ]
+        
+        filtered, count = filter_by_language(articles, "en")
+        
+        assert len(filtered) == 1
+        assert count == 1
+
+
+class TestFuzzyDeduplication:
+    """Tests for fuzzy title matching."""
+    
+    def test_exact_duplicate(self):
+        """Test that exact duplicates are detected."""
+        from app.data_manager import is_fuzzy_duplicate
+        
+        existing = ["Copper prices surge on supply concerns"]
+        new_title = "Copper prices surge on supply concerns"
+        
+        assert is_fuzzy_duplicate(new_title, existing, threshold=85) is True
+    
+    def test_similar_titles(self):
+        """Test that similar titles are detected."""
+        from app.data_manager import is_fuzzy_duplicate
+        
+        existing = ["Copper prices surge on supply concerns"]
+        new_title = "Copper prices rise on supply concerns"  # Similar
+        
+        # Should be detected as duplicate with default threshold
+        result = is_fuzzy_duplicate(new_title, existing, threshold=85)
+        assert result is True
+    
+    def test_different_titles(self):
+        """Test that different titles are not marked as duplicates."""
+        from app.data_manager import is_fuzzy_duplicate
+        
+        existing = ["Copper prices surge on supply concerns"]
+        new_title = "Gold reaches new all-time high"  # Different topic
+        
+        assert is_fuzzy_duplicate(new_title, existing, threshold=85) is False
+    
+    def test_empty_existing_titles(self):
+        """Test with no existing titles."""
+        from app.data_manager import is_fuzzy_duplicate
+        
+        existing = []
+        new_title = "Any title here"
+        
+        assert is_fuzzy_duplicate(new_title, existing, threshold=85) is False
+
+
+class TestRSSParsing:
+    """Tests for RSS feed parsing."""
+    
+    def test_rss_query_building(self):
+        """Test RSS query URL building."""
+        query = "copper OR copper price OR copper futures"
+        language = "en"
+        
+        # URL encoding
+        from urllib.parse import quote
+        encoded_query = quote(query)
+        
+        url = f"https://news.google.com/rss/search?q={encoded_query}&hl={language}&gl=US&ceid=US:en"
+        
+        assert "copper" in url
+        assert "hl=en" in url
+
+
+class TestPriceIngestion:
+    """Tests for price data ingestion."""
+    
+    def test_symbol_parsing(self):
+        """Test multi-symbol parsing."""
+        symbols_str = "HG=F,DX-Y.NYB,CL=F,FXI"
+        symbols = symbols_str.split(",")
+        
+        assert len(symbols) == 4
+        assert "HG=F" in symbols
+        assert "DX-Y.NYB" in symbols
+    
+    def test_lookback_calculation(self):
+        """Test lookback date calculation."""
+        lookback_days = 365
+        end_date = datetime.now(timezone.utc)
+        start_date = end_date - timedelta(days=lookback_days)
+        
+        delta = end_date - start_date
+        assert delta.days == lookback_days
+    
+    def test_price_bar_fields(self):
+        """Test that price bars have required fields."""
+        required_fields = ["date", "open", "high", "low", "close", "volume"]
+        
+        sample_bar = {
+            "date": datetime.now(),
+            "open": 4.0,
+            "high": 4.1,
+            "low": 3.9,
+            "close": 4.05,
+            "volume": 50000,
+        }
+        
+        for field in required_fields:
+            assert field in sample_bar
+
+
+class TestDatabaseUpsert:
+    """Tests for database upsert logic."""
+    
+    def test_upsert_key_generation(self):
+        """Test unique key generation for upsert."""
+        from app.utils import generate_dedup_key
+        
+        # Same URL should give same key
+        url = "https://example.com/article/123"
+        key1 = generate_dedup_key("Title 1", url)
+        key2 = generate_dedup_key("Title 2", url)
+        
+        # Keys based on URL should be consistent
+        # (depends on implementation - may include title or not)
+        assert isinstance(key1, str)
+        assert isinstance(key2, str)
+    
+    def test_date_normalization(self):
+        """Test date normalization for comparison."""
+        dt1 = datetime(2026, 1, 1, 10, 30, 0, tzinfo=timezone.utc)
+        dt2 = datetime(2026, 1, 1, 14, 45, 0, tzinfo=timezone.utc)
+        
+        # Same date, different time
+        date1 = dt1.date()
+        date2 = dt2.date()
+        
+        assert date1 == date2
+
+
+class TestDataValidation:
+    """Tests for data validation."""
+    
+    def test_price_validation(self):
+        """Test that prices are positive."""
+        prices = [4.0, 4.1, 4.05, 3.95]
+        
+        assert all(p > 0 for p in prices)
+    
+    def test_volume_validation(self):
+        """Test that volume is non-negative."""
+        volumes = [50000, 0, 100000]
+        
+        assert all(v >= 0 for v in volumes)
+    
+    def test_date_validation(self):
+        """Test date is not in future."""
+        from datetime import datetime, timezone
+        
+        test_date = datetime(2025, 1, 1, tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        
+        # For historical data, date should be in past or present
+        assert test_date <= now or True  # Flexible for test dates
+    
+    def test_sentiment_score_range(self):
+        """Test that sentiment scores are in valid range."""
+        scores = [0.5, -0.3, 0.8, -0.9, 0.0]
+        
+        assert all(-1 <= s <= 1 for s in scores)