Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/quant_platform_kit/risk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@
DEFAULT_TARGET_VOL_ANNUAL,
CapitalRiskEnvelope,
LeverageProductCap,
apply_combined_scale_to_targets,
apply_envelope_to_sized_weight,
evaluate_capital_risk_envelope,
)
Expand Down Expand Up @@ -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",
Expand Down
44 changes: 44 additions & 0 deletions src/quant_platform_kit/risk/capital_risk_envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
24 changes: 24 additions & 0 deletions tests/test_capital_risk_envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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()
Loading