From b031d2f9b107dae7dd5dd3b7b98a92aab39fa17f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:38:57 +0800 Subject: [PATCH 1/2] feat: add typed position sizing constraint engine Co-Authored-By: Codex --- src/quant_platform_kit/position_sizing_v1.py | 101 +++++++++++++++++++ tests/test_position_sizing_v1.py | 28 +++++ 2 files changed, 129 insertions(+) create mode 100644 src/quant_platform_kit/position_sizing_v1.py create mode 100644 tests/test_position_sizing_v1.py diff --git a/src/quant_platform_kit/position_sizing_v1.py b/src/quant_platform_kit/position_sizing_v1.py new file mode 100644 index 0000000..290212e --- /dev/null +++ b/src/quant_platform_kit/position_sizing_v1.py @@ -0,0 +1,101 @@ +"""Pure, research-only typed position sizing constraints (v1).""" +from __future__ import annotations +import hashlib +import json +import math +import re +from dataclasses import asdict,dataclass +from typing import Mapping + +class PositionSizingContractError(ValueError): pass +_DATE=re.compile(r"^\d{4}-\d{2}-\d{2}$") + +def _finite(value: object, field: str, *, low: float|None=None, high: float|None=None) -> float: + if isinstance(value,bool) or not isinstance(value,(int,float)) or not math.isfinite(float(value)): raise PositionSizingContractError(f"invalid {field}") + result=float(value) + if low is not None and resulthigh: raise PositionSizingContractError(f"invalid {field}") + return result + +def _text(value: object, field: str) -> str: + if not isinstance(value,str) or not value or len(value)>256 or not value.isprintable(): raise PositionSizingContractError(f"invalid {field}") + return value + +def _date(value: object, field: str) -> str: + value=_text(value,field) + if not _DATE.fullmatch(value): raise PositionSizingContractError(f"invalid {field}") + try: + from datetime import date; parsed=date.fromisoformat(value) + except ValueError: raise PositionSizingContractError(f"invalid {field}") from None + if parsed.isoformat()!=value: raise PositionSizingContractError(f"invalid {field}") + return value + +def _canonical(value: object) -> bytes: + try:return json.dumps(value,sort_keys=True,separators=(",",":"),ensure_ascii=True,allow_nan=False).encode("ascii") + except (TypeError,ValueError,UnicodeError): raise PositionSizingContractError("invalid canonical sizing") from None + +@dataclass(frozen=True,slots=True) +class StrategyTarget: + symbol: str; raw_weight: float + def __post_init__(self): + _text(self.symbol,"symbol"); _finite(self.raw_weight,"raw_weight",low=-1.0,high=1.0) + +@dataclass(frozen=True,slots=True) +class EvidenceRef: + package_id: str; scope: str; as_of: str; valid_until: str; digest: str; sample_count: int + def __post_init__(self): + _text(self.package_id,"package_id"); _text(self.scope,"scope"); a=_date(self.as_of,"as_of"); v=_date(self.valid_until,"valid_until") + if v PositionSizingPlan: + as_of=_date(as_of,"as_of") + if not isinstance(targets,tuple) or not isinstance(evidence,tuple) or not isinstance(caps,tuple): raise PositionSizingContractError("invalid immutable inputs") + if not evidence or any(e.as_of!=as_of or e.valid_untilpolicy.total_exposure_cap and total>0: + scale=policy.total_exposure_cap/total + built=[SizedTarget(x.symbol,x.raw_weight,x.final_weight*scale,x.caps,"total_exposure") for x in built] + payload={"as_of":as_of,"targets":[asdict(x) for x in built]} + dig=hashlib.sha256(_canonical(payload)).hexdigest(); ed=hashlib.sha256(_canonical([asdict(e) for e in evidence])).hexdigest(); pd=hashlib.sha256(_canonical(asdict(policy))).hexdigest() + return PositionSizingPlan(as_of,"ZERO_RISK" if risk_route in {"risk_off","blocked"} else "SIZED",tuple(built),ed,pd,dig,"risk_off_or_blocked" if risk_route in {"risk_off","blocked"} else None) diff --git a/tests/test_position_sizing_v1.py b/tests/test_position_sizing_v1.py new file mode 100644 index 0000000..bb54dfe --- /dev/null +++ b/tests/test_position_sizing_v1.py @@ -0,0 +1,28 @@ +import math +import pytest +from quant_platform_kit.position_sizing_v1 import ( + EvidenceRef, PositionSizingContractError, SizingAuthorization, SizingPolicy, + StrategyTarget, SymbolCaps, build_position_sizing_plan, +) + +def inputs(k=0.25): + return dict(as_of='2025-01-02',targets=(StrategyTarget('SOXL',.8),StrategyTarget('TQQQ',.4)),evidence=(EvidenceRef('e1','OOS','2025-01-02','2025-02-01','d1',100),),caps=(SymbolCaps('SOXL',.6,.5,.4,.3),SymbolCaps('TQQQ',.5,.5,.4,.3)),authorization=SizingAuthorization(True,'automation_approved',.8,'2025-02-01'),policy=SizingPolicy(k, .7,.1,1.0)) + +def test_fractional_caps_and_never_increase(): + p=build_position_sizing_plan(**inputs(.25)); assert p.status=='SIZED'; assert all(abs(x.final_weight)<=abs(x.raw_weight) for x in p.targets); assert p.targets[0].final_weight==pytest.approx(.12) + assert build_position_sizing_plan(**inputs(.5)).targets[0].final_weight==pytest.approx(.24) + +def test_total_scaling_and_risk_off(): + p=build_position_sizing_plan(**inputs()); assert sum(abs(x.final_weight) for x in p.targets)<=.7 + z=build_position_sizing_plan(**inputs(),risk_route='risk_off'); assert z.status=='ZERO_RISK' and all(x.final_weight==0 for x in z.targets) + +def test_evidence_auth_and_determinism(): + p=build_position_sizing_plan(**inputs()); q=build_position_sizing_plan(**inputs()); assert p==q and p.input_digest + with pytest.raises(PositionSizingContractError): build_position_sizing_plan(**{**inputs(),'as_of':'2025-03-01'}) + with pytest.raises(PositionSizingContractError): build_position_sizing_plan(**{**inputs(),'authorization':SizingAuthorization(False,'automation_approved',.8,'2025-02-01')}) + +def test_invalid_values_and_boundary(): + with pytest.raises(PositionSizingContractError): StrategyTarget('X',math.nan) + with pytest.raises(PositionSizingContractError): SizingPolicy(.3) + with pytest.raises(PositionSizingContractError): EvidenceRef('e','OOS','2025-01-02','2024-01-01','d',1) + with pytest.raises(PositionSizingContractError): build_position_sizing_plan(**{**inputs(),'targets':(StrategyTarget('SOXL',.1),)}) From 868649195753fdbc054908a5343683d09dae0156 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:50:23 +0800 Subject: [PATCH 2/2] fix: bind sizing digest and cash reserve Co-Authored-By: Codex --- src/quant_platform_kit/position_sizing_v1.py | 18 +++++++++++++----- tests/test_position_sizing_v1.py | 7 +++++++ 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/src/quant_platform_kit/position_sizing_v1.py b/src/quant_platform_kit/position_sizing_v1.py index 290212e..4dba8df 100644 --- a/src/quant_platform_kit/position_sizing_v1.py +++ b/src/quant_platform_kit/position_sizing_v1.py @@ -86,6 +86,15 @@ def build_position_sizing_plan(*, as_of: str, targets: tuple[StrategyTarget,...] if risk_route not in {"no_action","watch","opportunity_watch","risk_reduced","risk_off","blocked"}: raise PositionSizingContractError("invalid risk route") names=[t.symbol for t in targets]; capmap={c.symbol:c for c in caps} if len(set(names))!=len(names) or len(capmap)!=len(caps) or set(names)!=set(capmap): raise PositionSizingContractError("symbol mismatch") + input_payload={ + "as_of":as_of, + "targets":[asdict(x) for x in sorted(targets,key=lambda x:x.symbol)], + "evidence":[asdict(x) for x in sorted(evidence,key=lambda x:(x.package_id,x.scope,x.digest))], + "caps":[asdict(x) for x in sorted(caps,key=lambda x:x.symbol)], + "authorization":asdict(authorization),"policy":asdict(policy),"risk_route":risk_route, + } + input_digest=hashlib.sha256(_canonical(input_payload)).hexdigest() + reserve_aware_cap=max(0.0,policy.total_exposure_cap-policy.cash_reserve) scalar=min(policy.risk_scalar,authorization.bounded_budget,0.0 if risk_route in {"risk_off","blocked"} else 1.0) built=[] for t in targets: @@ -93,9 +102,8 @@ def build_position_sizing_plan(*, as_of: str, targets: tuple[StrategyTarget,...] if final==0: binding="zero_cap" built.append(SizedTarget(t.symbol,t.raw_weight,math.copysign(final,t.raw_weight) if final else 0.0,(("kelly",abs(t.raw_weight),k),("volatility",k,c.volatility_cap),("correlation",k,c.correlation_cap),("liquidity",k,c.liquidity_cap)),binding)) total=sum(abs(x.final_weight) for x in built) - if total>policy.total_exposure_cap and total>0: - scale=policy.total_exposure_cap/total + if total>reserve_aware_cap and total>0: + scale=reserve_aware_cap/total built=[SizedTarget(x.symbol,x.raw_weight,x.final_weight*scale,x.caps,"total_exposure") for x in built] - payload={"as_of":as_of,"targets":[asdict(x) for x in built]} - dig=hashlib.sha256(_canonical(payload)).hexdigest(); ed=hashlib.sha256(_canonical([asdict(e) for e in evidence])).hexdigest(); pd=hashlib.sha256(_canonical(asdict(policy))).hexdigest() - return PositionSizingPlan(as_of,"ZERO_RISK" if risk_route in {"risk_off","blocked"} else "SIZED",tuple(built),ed,pd,dig,"risk_off_or_blocked" if risk_route in {"risk_off","blocked"} else None) + ed=hashlib.sha256(_canonical(input_payload["evidence"])).hexdigest(); pd=hashlib.sha256(_canonical(asdict(policy))).hexdigest() + return PositionSizingPlan(as_of,"ZERO_RISK" if risk_route in {"risk_off","blocked"} else "SIZED",tuple(built),ed,pd,input_digest,"risk_off_or_blocked" if risk_route in {"risk_off","blocked"} else None) diff --git a/tests/test_position_sizing_v1.py b/tests/test_position_sizing_v1.py index bb54dfe..69fa8b4 100644 --- a/tests/test_position_sizing_v1.py +++ b/tests/test_position_sizing_v1.py @@ -16,6 +16,13 @@ def test_total_scaling_and_risk_off(): p=build_position_sizing_plan(**inputs()); assert sum(abs(x.final_weight) for x in p.targets)<=.7 z=build_position_sizing_plan(**inputs(),risk_route='risk_off'); assert z.status=='ZERO_RISK' and all(x.final_weight==0 for x in z.targets) +def test_cash_reserve_reduces_budget_and_digest_binds_inputs(): + base=inputs(); p=build_position_sizing_plan(**base) + reserved=build_position_sizing_plan(**{**base,'policy':SizingPolicy(.25,.7,.6,1.0)}) + assert sum(abs(x.final_weight) for x in reserved.targets)<=.1000001 + assert reserved.input_digest!=p.input_digest + assert build_position_sizing_plan(**{**base,'risk_route':'blocked'}).targets[0].final_weight==0 + def test_evidence_auth_and_determinism(): p=build_position_sizing_plan(**inputs()); q=build_position_sizing_plan(**inputs()); assert p==q and p.input_digest with pytest.raises(PositionSizingContractError): build_position_sizing_plan(**{**inputs(),'as_of':'2025-03-01'})