diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index 3337bbe..1e6cdd7 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/src/quant_platform_kit/risk/__init__.py @@ -69,6 +69,7 @@ DEFAULT_TARGET_VOL_ANNUAL, CapitalRiskEnvelope, LeverageProductCap, + apply_combined_scale_to_targets, apply_envelope_to_sized_weight, evaluate_capital_risk_envelope, ) @@ -152,6 +153,7 @@ "CapitalRiskEnvelope", "LeverageProductCap", "apply_envelope_to_sized_weight", + "apply_combined_scale_to_targets", "evaluate_capital_risk_envelope", "ReconciliationEquitySummary", "build_injected_snapshot_from_equity_summary", diff --git a/src/quant_platform_kit/risk/capital_risk_envelope.py b/src/quant_platform_kit/risk/capital_risk_envelope.py index d3bb750..dee1c05 100644 --- a/src/quant_platform_kit/risk/capital_risk_envelope.py +++ b/src/quant_platform_kit/risk/capital_risk_envelope.py @@ -389,12 +389,56 @@ def apply_envelope_to_sized_weight( return min(float(sized_weight), 1.0, out) +def apply_combined_scale_to_targets( + targets: Mapping[str, Any] | None, + combined_scale: float | None, +) -> dict[str, float]: + """Shrink target values/weights by ``combined_scale``; never invent risk. + + Live wiring rule (distinct from ``apply_envelope_to_sized_weight``): + - ``combined_scale is None`` / non-finite / negative → **omit** (return a + copy of finite non-negative inputs unchanged). Do not fail-closed to zero. + - Valid scale → each value becomes ``min(original, original * clamp(scale,0,1))``. + - Unknown / negative input values are dropped (not promoted to targets). + """ + + raw = dict(targets or {}) + try: + if combined_scale is None or isinstance(combined_scale, bool): + scale: float | None = None + else: + number = float(combined_scale) + scale = None if (not math.isfinite(number) or number < 0.0) else _clamp_unit(number) + except (TypeError, ValueError): + scale = None + + out: dict[str, float] = {} + for key, value in raw.items(): + symbol = str(key or "").strip() + if not symbol: + continue + try: + if isinstance(value, bool): + continue + amount = float(value) + except (TypeError, ValueError): + continue + if not math.isfinite(amount) or amount < 0.0: + continue + if scale is None: + out[symbol] = amount + else: + out[symbol] = min(amount, amount * scale) + return out + + __all__ = [ "DEFAULT_TARGET_VOL_ANNUAL", "AccountCapitalEnvelopeSummary", "CapitalRiskEnvelope", "LeverageProductCap", "MultiAccountCapitalEnvelopeView", + "apply_combined_scale_to_targets", "apply_envelope_to_sized_weight", "evaluate_capital_risk_envelope", "evaluate_multi_account_envelope_view", diff --git a/tests/test_capital_risk_envelope.py b/tests/test_capital_risk_envelope.py index 84234d8..3cc8084 100644 --- a/tests/test_capital_risk_envelope.py +++ b/tests/test_capital_risk_envelope.py @@ -7,6 +7,7 @@ from quant_platform_kit.risk.capital_risk_envelope import ( DEFAULT_TARGET_VOL_ANNUAL, CapitalRiskEnvelope, + apply_combined_scale_to_targets, apply_envelope_to_sized_weight, evaluate_multi_account_envelope_view, evaluate_capital_risk_envelope, @@ -213,5 +214,28 @@ def test_result_is_capital_risk_envelope_dataclass(self) -> None: self.assertIsInstance(env, CapitalRiskEnvelope) +class ApplyCombinedScaleToTargetsTests(unittest.TestCase): + def test_none_scale_omits_without_zeroing(self) -> None: + targets = {"SOXL": 0.60, "SOXX": 0.20} + self.assertEqual(apply_combined_scale_to_targets(targets, None), targets) + + def test_invalid_scale_omits(self) -> None: + targets = {"SOXL": 1000.0} + self.assertEqual(apply_combined_scale_to_targets(targets, float("nan")), targets) + self.assertEqual(apply_combined_scale_to_targets(targets, -0.1), targets) + + def test_valid_scale_only_shrinks(self) -> None: + targets = {"SOXL": 1000.0, "BOXX": 200.0} + scaled = apply_combined_scale_to_targets(targets, 0.85) + self.assertAlmostEqual(scaled["SOXL"], 850.0) + self.assertAlmostEqual(scaled["BOXX"], 170.0) + self.assertLessEqual(scaled["SOXL"], targets["SOXL"]) + + def test_scale_above_one_clamped(self) -> None: + targets = {"QQQ": 0.50} + scaled = apply_combined_scale_to_targets(targets, 1.5) + self.assertAlmostEqual(scaled["QQQ"], 0.50) + + if __name__ == "__main__": unittest.main()