156 lines
4.9 KiB
Python
156 lines
4.9 KiB
Python
import unittest
|
|
from datetime import date
|
|
|
|
from quant60.config import FeeConfig
|
|
from quant60.costs import (
|
|
CostModelParameters,
|
|
DatedFeeRule,
|
|
VersionedFeeSchedule,
|
|
forecast_trade_cost,
|
|
)
|
|
from quant60.domain import Side
|
|
from quant60.optimizer import (
|
|
OptimizationError,
|
|
PortfolioCandidate,
|
|
PortfolioConstraints,
|
|
optimize_cvxpy,
|
|
optimize_deterministic,
|
|
)
|
|
from quant60.universe import UniverseState
|
|
|
|
|
|
class CostOptimizerTests(unittest.TestCase):
|
|
def test_fee_schedule_switches_by_effective_date(self):
|
|
first = DatedFeeRule(
|
|
date(2024, 1, 1),
|
|
date(2024, 6, 30),
|
|
FeeConfig(stamp_duty_rate=0.001),
|
|
"old",
|
|
)
|
|
second = DatedFeeRule(
|
|
date(2024, 7, 1),
|
|
None,
|
|
FeeConfig(stamp_duty_rate=0.0005),
|
|
"new",
|
|
)
|
|
schedule = VersionedFeeSchedule([second, first])
|
|
self.assertEqual(schedule.at(date(2024, 6, 30)).version, "old")
|
|
self.assertEqual(schedule.at(date(2024, 7, 1)).version, "new")
|
|
|
|
def test_exact_minimum_fee_and_capacity_are_visible(self):
|
|
rule = DatedFeeRule(
|
|
date(2024, 1, 1),
|
|
None,
|
|
FeeConfig(
|
|
commission_rate=0.0002,
|
|
minimum_commission=5,
|
|
stamp_duty_rate=0.0005,
|
|
transfer_fee_rate=0.00001,
|
|
),
|
|
"account-v1",
|
|
)
|
|
forecast = forecast_trade_cost(
|
|
symbol="600000.SH",
|
|
side=Side.BUY,
|
|
quantity=100,
|
|
price=10,
|
|
adv_quantity=500,
|
|
daily_volatility=0.02,
|
|
fee_rule=rule,
|
|
parameters=CostModelParameters(max_participation=0.10),
|
|
)
|
|
self.assertGreaterEqual(forecast.explicit_fees, 5)
|
|
self.assertTrue(forecast.capacity_breached)
|
|
self.assertLess(forecast.low, forecast.base)
|
|
self.assertLess(forecast.base, forecast.high)
|
|
|
|
def _candidates(self):
|
|
return [
|
|
PortfolioCandidate(
|
|
"600000.SH",
|
|
score=0.04,
|
|
variance=0.04,
|
|
current_weight=0.10,
|
|
cost_bps=10,
|
|
max_adv_weight=0.30,
|
|
industry="bank",
|
|
),
|
|
PortfolioCandidate(
|
|
"000001.SZ",
|
|
score=0.03,
|
|
variance=0.03,
|
|
current_weight=0.10,
|
|
cost_bps=12,
|
|
max_adv_weight=0.30,
|
|
industry="bank",
|
|
),
|
|
PortfolioCandidate(
|
|
"300750.SZ",
|
|
score=0.02,
|
|
variance=0.05,
|
|
current_weight=0.05,
|
|
cost_bps=15,
|
|
max_adv_weight=0.30,
|
|
industry="industry",
|
|
state=UniverseState.FROZEN,
|
|
),
|
|
PortfolioCandidate(
|
|
"000333.SZ",
|
|
score=-0.01,
|
|
variance=0.04,
|
|
current_weight=0.10,
|
|
cost_bps=10,
|
|
max_adv_weight=0.30,
|
|
industry="consumer",
|
|
state=UniverseState.SELL_ONLY,
|
|
),
|
|
]
|
|
|
|
def test_deterministic_optimizer_honors_frozen_caps_and_turnover(self):
|
|
constraints = PortfolioConstraints(
|
|
gross_target=0.70,
|
|
single_name_cap=0.30,
|
|
industry_cap=0.45,
|
|
one_way_turnover_cap=0.10,
|
|
)
|
|
result = optimize_deterministic(self._candidates(), constraints)
|
|
self.assertAlmostEqual(result.weights["300750.XSHE"], 0.05)
|
|
self.assertLessEqual(result.weights["000333.XSHE"], 0.10)
|
|
self.assertLessEqual(result.one_way_turnover, 0.10 + 1e-9)
|
|
self.assertLessEqual(result.gross_weight, 0.70 + 1e-9)
|
|
self.assertEqual(result.solver, "deterministic_reference")
|
|
|
|
def test_held_excluded_candidate_fails_closed(self):
|
|
candidate = PortfolioCandidate(
|
|
"600000.SH",
|
|
score=0,
|
|
variance=0.04,
|
|
current_weight=0.1,
|
|
cost_bps=1,
|
|
max_adv_weight=0.2,
|
|
industry="bank",
|
|
state=UniverseState.EXCLUDED,
|
|
)
|
|
with self.assertRaisesRegex(OptimizationError, "cannot disappear"):
|
|
optimize_deterministic([candidate])
|
|
|
|
def test_cvxpy_path_when_dependency_is_available(self):
|
|
constraints = PortfolioConstraints(
|
|
gross_target=0.70,
|
|
single_name_cap=0.30,
|
|
industry_cap=0.45,
|
|
one_way_turnover_cap=0.20,
|
|
)
|
|
try:
|
|
result = optimize_cvxpy(self._candidates(), constraints)
|
|
except OptimizationError as exc:
|
|
if "CVXPY is unavailable" in str(exc):
|
|
self.skipTest(str(exc))
|
|
raise
|
|
self.assertIn(result.status, {"OPTIMAL", "OPTIMAL_INACCURATE"})
|
|
self.assertLessEqual(result.one_way_turnover, 0.20 + 1e-6)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|