250 lines
8.3 KiB
Python
250 lines
8.3 KiB
Python
|
"""
|
||
|
|
Strategy consistency tests
|
||
|
|
"""
|
||
|
|
import pytest
|
||
|
|
from datetime import date
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.strategies.composite.multi_factor import MultiFactorStrategy
|
||
|
|
from app.strategies.composite.magic_formula import MagicFormulaStrategy
|
||
|
|
from app.strategies.composite.super_quality import SuperQualityStrategy
|
||
|
|
from app.strategies.factors.momentum import MomentumStrategy
|
||
|
|
from app.strategies.factors.f_score import FScoreStrategy
|
||
|
|
from app.strategies.factors.value import ValueStrategy
|
||
|
|
from app.strategies.factors.quality import QualityStrategy
|
||
|
|
from app.strategies.factors.all_value import AllValueStrategy
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.unit
|
||
|
|
class TestStrategyInterface:
|
||
|
|
"""Test strategy interface implementation"""
|
||
|
|
|
||
|
|
def test_multi_factor_strategy_interface(self):
|
||
|
|
"""Test MultiFactorStrategy implements BaseStrategy"""
|
||
|
|
strategy = MultiFactorStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "multi_factor"
|
||
|
|
|
||
|
|
def test_magic_formula_strategy_interface(self):
|
||
|
|
"""Test MagicFormulaStrategy implements BaseStrategy"""
|
||
|
|
strategy = MagicFormulaStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "magic_formula"
|
||
|
|
|
||
|
|
def test_super_quality_strategy_interface(self):
|
||
|
|
"""Test SuperQualityStrategy implements BaseStrategy"""
|
||
|
|
strategy = SuperQualityStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "super_quality"
|
||
|
|
|
||
|
|
def test_momentum_strategy_interface(self):
|
||
|
|
"""Test MomentumStrategy implements BaseStrategy"""
|
||
|
|
strategy = MomentumStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "momentum"
|
||
|
|
|
||
|
|
def test_f_score_strategy_interface(self):
|
||
|
|
"""Test FScoreStrategy implements BaseStrategy"""
|
||
|
|
strategy = FScoreStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "f_score"
|
||
|
|
|
||
|
|
def test_value_strategy_interface(self):
|
||
|
|
"""Test ValueStrategy implements BaseStrategy"""
|
||
|
|
strategy = ValueStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "value"
|
||
|
|
|
||
|
|
def test_quality_strategy_interface(self):
|
||
|
|
"""Test QualityStrategy implements BaseStrategy"""
|
||
|
|
strategy = QualityStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "quality"
|
||
|
|
|
||
|
|
def test_all_value_strategy_interface(self):
|
||
|
|
"""Test AllValueStrategy implements BaseStrategy"""
|
||
|
|
strategy = AllValueStrategy(config={"count": 20})
|
||
|
|
|
||
|
|
assert hasattr(strategy, "select_stocks")
|
||
|
|
assert hasattr(strategy, "get_prices")
|
||
|
|
assert strategy.name == "all_value"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.integration
|
||
|
|
@pytest.mark.slow
|
||
|
|
class TestStrategyExecution:
|
||
|
|
"""Test strategy execution with sample data"""
|
||
|
|
|
||
|
|
def test_multi_factor_select_stocks(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test MultiFactorStrategy stock selection"""
|
||
|
|
strategy = MultiFactorStrategy(config={"count": 3})
|
||
|
|
rebal_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
# Note: May fail if insufficient data, that's expected
|
||
|
|
try:
|
||
|
|
selected_stocks = strategy.select_stocks(rebal_date, db_session)
|
||
|
|
|
||
|
|
# Should return list of tickers
|
||
|
|
assert isinstance(selected_stocks, list)
|
||
|
|
assert len(selected_stocks) <= 3
|
||
|
|
|
||
|
|
for ticker in selected_stocks:
|
||
|
|
assert isinstance(ticker, str)
|
||
|
|
assert len(ticker) == 6
|
||
|
|
except Exception as e:
|
||
|
|
# Insufficient data is acceptable for test
|
||
|
|
pytest.skip(f"Insufficient data for strategy execution: {e}")
|
||
|
|
|
||
|
|
def test_momentum_select_stocks(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test MomentumStrategy stock selection"""
|
||
|
|
strategy = MomentumStrategy(config={"count": 3})
|
||
|
|
rebal_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
try:
|
||
|
|
selected_stocks = strategy.select_stocks(rebal_date, db_session)
|
||
|
|
|
||
|
|
assert isinstance(selected_stocks, list)
|
||
|
|
assert len(selected_stocks) <= 3
|
||
|
|
except Exception as e:
|
||
|
|
pytest.skip(f"Insufficient data for strategy execution: {e}")
|
||
|
|
|
||
|
|
def test_value_select_stocks(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test ValueStrategy stock selection"""
|
||
|
|
strategy = ValueStrategy(config={"count": 3})
|
||
|
|
rebal_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
try:
|
||
|
|
selected_stocks = strategy.select_stocks(rebal_date, db_session)
|
||
|
|
|
||
|
|
assert isinstance(selected_stocks, list)
|
||
|
|
assert len(selected_stocks) <= 3
|
||
|
|
|
||
|
|
for ticker in selected_stocks:
|
||
|
|
assert isinstance(ticker, str)
|
||
|
|
assert len(ticker) == 6
|
||
|
|
except Exception as e:
|
||
|
|
pytest.skip(f"Insufficient data for strategy execution: {e}")
|
||
|
|
|
||
|
|
def test_quality_select_stocks(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test QualityStrategy stock selection"""
|
||
|
|
strategy = QualityStrategy(config={"count": 3})
|
||
|
|
rebal_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
try:
|
||
|
|
selected_stocks = strategy.select_stocks(rebal_date, db_session)
|
||
|
|
|
||
|
|
assert isinstance(selected_stocks, list)
|
||
|
|
assert len(selected_stocks) <= 3
|
||
|
|
|
||
|
|
for ticker in selected_stocks:
|
||
|
|
assert isinstance(ticker, str)
|
||
|
|
assert len(ticker) == 6
|
||
|
|
except Exception as e:
|
||
|
|
pytest.skip(f"Insufficient data for strategy execution: {e}")
|
||
|
|
|
||
|
|
def test_all_value_select_stocks(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test AllValueStrategy stock selection"""
|
||
|
|
strategy = AllValueStrategy(config={"count": 3})
|
||
|
|
rebal_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
try:
|
||
|
|
selected_stocks = strategy.select_stocks(rebal_date, db_session)
|
||
|
|
|
||
|
|
assert isinstance(selected_stocks, list)
|
||
|
|
assert len(selected_stocks) <= 3
|
||
|
|
|
||
|
|
for ticker in selected_stocks:
|
||
|
|
assert isinstance(ticker, str)
|
||
|
|
assert len(ticker) == 6
|
||
|
|
except Exception as e:
|
||
|
|
pytest.skip(f"Insufficient data for strategy execution: {e}")
|
||
|
|
|
||
|
|
def test_strategy_get_prices(
|
||
|
|
self,
|
||
|
|
db_session: Session,
|
||
|
|
sample_assets,
|
||
|
|
sample_price_data
|
||
|
|
):
|
||
|
|
"""Test strategy price retrieval"""
|
||
|
|
strategy = MultiFactorStrategy(config={"count": 3})
|
||
|
|
tickers = ["005930", "000660", "035420"]
|
||
|
|
price_date = date(2023, 1, 15)
|
||
|
|
|
||
|
|
prices = strategy.get_prices(tickers, price_date, db_session)
|
||
|
|
|
||
|
|
# Should return dict of prices
|
||
|
|
assert isinstance(prices, dict)
|
||
|
|
|
||
|
|
# May not have all prices if data is incomplete
|
||
|
|
for ticker, price in prices.items():
|
||
|
|
assert ticker in tickers
|
||
|
|
assert price > 0
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.integration
|
||
|
|
class TestStrategyConfiguration:
|
||
|
|
"""Test strategy configuration handling"""
|
||
|
|
|
||
|
|
def test_strategy_default_config(self):
|
||
|
|
"""Test strategy with default configuration"""
|
||
|
|
strategy = MultiFactorStrategy(config={})
|
||
|
|
|
||
|
|
# Should use default count
|
||
|
|
assert "count" in strategy.config or hasattr(strategy, "count")
|
||
|
|
|
||
|
|
def test_strategy_custom_count(self):
|
||
|
|
"""Test strategy with custom count"""
|
||
|
|
strategy = MultiFactorStrategy(config={"count": 50})
|
||
|
|
|
||
|
|
assert strategy.config["count"] == 50
|
||
|
|
|
||
|
|
def test_strategy_invalid_config(self):
|
||
|
|
"""Test strategy with invalid configuration"""
|
||
|
|
# Should handle gracefully or raise appropriate error
|
||
|
|
try:
|
||
|
|
strategy = MultiFactorStrategy(config={"count": -1})
|
||
|
|
# If it doesn't raise, it should handle gracefully
|
||
|
|
assert True
|
||
|
|
except ValueError:
|
||
|
|
# Expected for negative count
|
||
|
|
assert True
|