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
35 changes: 31 additions & 4 deletions src/quant_platform_kit/position_sizing.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,14 @@ class KellyResult:


_APPROVED_BOOTSTRAP_MANDATE = "bootstrap_small_account_v2"
_TQQQ_ETF_ONLY_RESEARCH_MANDATE = "tqqq_etf_only_research_v1"
_BOOTSTRAP_LOSS_BUDGET_CAP = 0.01
_BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP = 0.50
_BOOTSTRAP_NOMINAL_CAPS = {1: 0.50, 2: 0.25, 3: 0.15}
_TQQQ_ETF_ONLY_PRODUCTS = {
"TQQQ": (3, 0.15, 0.45),
"BOXX": (1, 0.50, 0.50),
}


def _weight_mapping(value: object, *, allow_empty: bool) -> dict[str, float] | None:
Expand Down Expand Up @@ -140,6 +145,7 @@ def validate_reduce_only_normalization(
product_leverage_factors: Mapping[str, int],
effective_exposure_cap: float,
observed_effective_exposure: float,
cash_only: bool = False,
) -> bool:
"""Validate one explicit transition from an over-cap origin toward cash."""
origin = _weight_mapping(origin_weights, allow_empty=False)
Expand All @@ -153,6 +159,7 @@ def validate_reduce_only_normalization(
or not isinstance(effective_exposure_cap, (int, float))
or isinstance(observed_effective_exposure, bool)
or not isinstance(observed_effective_exposure, (int, float))
or not isinstance(cash_only, bool)
):
return False
cap = float(effective_exposure_cap)
Expand All @@ -178,6 +185,8 @@ def validate_reduce_only_normalization(
target_active = {symbol for symbol, weight in target.items() if weight > 0.0}
if not origin_active or not target_active.issubset(origin_active):
return False
if cash_only and target_active:
return False
if any(target.get(symbol, 0.0) > origin[symbol] + 1e-9 for symbol in origin):
return False
if any(
Expand All @@ -199,6 +208,7 @@ def validate_reduce_only_normalization(
def risk_budgeted_target_weight(
*,
risk_mandate_id: str | None = None,
product_symbol: str | None = None,
account_equity: float | None = None,
risk_fraction: float | None = None,
stop_loss_distance: float | None = None,
Expand All @@ -209,9 +219,9 @@ def risk_budgeted_target_weight(
) -> float:
"""Return a fail-closed single-account target weight.

``bootstrap_small_account_v2`` permits one ETF position with its approved
product cap. Without that mandate, the legacy 10% and unlevered fallback
applies. This helper does not allocate across strategies.
Approved mandates permit one ETF position with their product caps. Without
a mandate, the legacy 10% and unlevered fallback applies. This pure helper
sizes a target; it does not authorize a product or execution.
"""
numeric_inputs = (
account_equity,
Expand Down Expand Up @@ -256,18 +266,35 @@ def risk_budgeted_target_weight(
nominal_cap = _BOOTSTRAP_NOMINAL_CAPS.get(product_leverage_factor, 0.0)
if nominal_cap == 0.0:
return 0.0
elif risk_mandate_id == _TQQQ_ETF_ONLY_RESEARCH_MANDATE:
product = _TQQQ_ETF_ONLY_PRODUCTS.get(product_symbol or "")
if (
product is None
or product_leverage_factor != product[0]
or risk_fraction != _BOOTSTRAP_LOSS_BUDGET_CAP
or stop_loss_distance != 0.05
or drawdown_scalar not in {0.0, 0.5, 1.0}
):
return 0.0
nominal_cap = product[1]
product_effective_cap = product[2]
else:
return 0.0

risk_weight = risk_fraction * drawdown_scalar / stop_loss_distance
if not math.isfinite(risk_weight):
return 0.0

effective_cap = (
product_effective_cap
if risk_mandate_id == _TQQQ_ETF_ONLY_RESEARCH_MANDATE
else _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP
)
return min(
risk_weight,
nominal_cap,
available_account_exposure,
_BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP / product_leverage_factor,
effective_cap / product_leverage_factor,
)


Expand Down
16 changes: 16 additions & 0 deletions src/quant_platform_kit/risk/contracts.py
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,14 @@ class RiskGateAssessment:
proposed_effective_exposure: float | None
outcome: str
reason_codes: tuple[str, ...]
execution_authorized: bool = False
stop_loss_distance: float | None = None
stop_intent_ready: bool | None = None
strategy_breaker_triggered: bool | None = None
account_breaker_triggered: bool | None = None
account_drawdown_fraction: float | None = None
drawdown_scalar: float | None = None
risk_control_state_digest_sha256: str | None = None
assessment_sha256: str = field(init=False)

def __post_init__(self) -> None:
Expand All @@ -245,6 +253,14 @@ def __post_init__(self) -> None:
"proposed_effective_exposure": self.proposed_effective_exposure,
"outcome": self.outcome,
"reason_codes": self.reason_codes,
"execution_authorized": self.execution_authorized,
"stop_loss_distance": self.stop_loss_distance,
"stop_intent_ready": self.stop_intent_ready,
"strategy_breaker_triggered": self.strategy_breaker_triggered,
"account_breaker_triggered": self.account_breaker_triggered,
"account_drawdown_fraction": self.account_drawdown_fraction,
"drawdown_scalar": self.drawdown_scalar,
"risk_control_state_digest_sha256": self.risk_control_state_digest_sha256,
}
encoded = json.dumps(
payload,
Expand Down
Loading