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
10 changes: 10 additions & 0 deletions config/baskets.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ baskets:
kr_diversified_hold:
name: "분산 대형주 buy&hold (저회전)"
enabled: true
# 관찰용 강등(docs/POCKET_TRACK_PLAN.md §6): 소액 전환으로 자본 증액 결정이 종결돼
# 배치율 미달(-21%p, 하이닉스 슬롯+절사)은 조치 대상이 아니다 — 상시 ATTENTION은
# 경보 피로만 남기고 다른 바스켓의 배치율 이탈을 가린다. 감시 사실상 해제(1.0).
monitoring:
deployment_tolerance: 1.0
rebalance:
trigger: drift # 분기 점검을 권장하나 drift로 큰 이탈만 교정
drift_threshold: 0.08 # 8%p 이상 이탈 시에만 리밸런싱(잦은 매매 억제)
Expand Down Expand Up @@ -95,6 +100,11 @@ baskets:
enabled: true
initial_capital: 300000 # 시작 자본 30만원 (paper — live 전환 시 실자본과 일치)
target_stock_weight: 0.5 # 주식 50% / 현금 50% — 낙폭 절반이 목적
# 소액 구조적 절사: ETF 1주(약 12.8만) 단위라 30만 시점 배치율이 -7.3%p 상시 미달
# (docs/POCKET_TRACK_PLAN.md §3 — 적립 누적 시 자연 수렴). 기본 허용 5%p면 매일
# 무의미한 ATTENTION이라 10%p로 완화 — 이걸 넘으면 절사가 아니라 진짜 이상이다.
monitoring:
deployment_tolerance: 0.10
rebalance:
trigger: drift
drift_threshold: 0.08
Expand Down
18 changes: 14 additions & 4 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1710,6 +1710,7 @@ def _design_fraction(cfg: dict) -> float:
worst_shortfall = None
worst_dep_ratio = None
worst_design = None
worst_tolerance = 0.05
session = get_session()
try:
for name in enabled_baskets:
Expand All @@ -1725,15 +1726,23 @@ def _design_fraction(cfg: dict) -> float:
# 배치율은 부가 신호 — 계산 실패(예: baskets.yaml에 float 불가한
# target_stock_weight 오타)가 핵심 신호인 결측/staleness 감지를
# 통째로 삼키지 않도록 자체 try로 격리한다.
# 바스켓별 허용 오차(monitoring.deployment_tolerance, 기본 5%p)를
# 반영해 '허용 초과분'이 가장 큰 바스켓을 고른다 — 원시 미달폭
# 최대로 고르면 관찰용 트랙(허용 완화)이 진짜 감시 대상을 가린다.
try:
if snap and snap.total_value and snap.total_value > 0:
cfg_b = baskets_cfg.get(name) or {}
dep_ratio = max(0.0, (snap.total_value - (snap.cash or 0)) / snap.total_value)
design = _design_fraction(baskets_cfg.get(name) or {})
shortfall = design - dep_ratio
if worst_shortfall is None or shortfall > worst_shortfall:
worst_shortfall = shortfall
design = _design_fraction(cfg_b)
tol_b = float(
(cfg_b.get("monitoring") or {}).get("deployment_tolerance", 0.05)
)
violation = (design - dep_ratio) - tol_b
if worst_shortfall is None or violation > worst_shortfall:
worst_shortfall = violation
worst_dep_ratio = dep_ratio
worst_design = design
worst_tolerance = tol_b
except Exception as dep_exc:
logger.debug("바스켓 '{}' 배치율 계산 생략: {}", name, dep_exc)
finally:
Expand All @@ -1749,6 +1758,7 @@ def _design_fraction(cfg: dict) -> float:
"today": date.today(),
"deployment_ratio": worst_dep_ratio,
"design_fraction": worst_design,
"deployment_tolerance": worst_tolerance,
}
except Exception as exc:
logger.warning("바스켓 운영 상태 조회 실패: {}", exc)
Expand Down
9 changes: 9 additions & 0 deletions tests/test_basket_rebalancer.py
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,15 @@ def test_pocket_basket_small_capital_invariants(self):
assert float(rb["max_turnover_ratio"]) >= tsw, "회전 한도 < 슬리브면 초기 매수 불가"
# ETF 단일 구성(1주 = 200종목 분산)
assert list(b["holdings"].keys()) == ["069500"]
# 소액 구조적 절사(1주 단위 -7.3%p) 허용 — 기본 5%p면 매일 무의미 ATTENTION
assert float(b["monitoring"]["deployment_tolerance"]) == 0.10

def test_observation_track_deployment_alarm_disabled(self):
"""관찰용 강등(kr_diversified_hold): 종결된 자본 결정의 잔상인 배치율 미달이
상시 ATTENTION으로 남아 다른 바스켓 감시를 가리지 않도록 허용 오차 해제."""
baskets = self._load()
b = baskets["kr_diversified_hold"]
assert float(b["monitoring"]["deployment_tolerance"]) >= 1.0

def test_all_basket_symbols_are_6digit_kr_codes(self):
baskets = self._load()
Expand Down
12 changes: 12 additions & 0 deletions tests/test_operator_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,18 @@ def test_deployment_none_is_ok(self):
out = self._summ(deployment_ratio=None, design_fraction=None)
assert out["verdict"] == "OK"

def test_per_basket_tolerance_relaxes_attention(self):
# 관찰용 강등 바스켓: 허용 오차 완화 시 큰 미달도 OK (경보 피로 방지)
out = self._summ(
deployment_ratio=0.59, design_fraction=0.80, deployment_tolerance=1.0,
)
assert out["verdict"] == "OK"
# 소액 구조적 절사(10%p 허용): -7.3%p는 OK, -12%p는 ATTENTION
ok = self._summ(deployment_ratio=0.427, design_fraction=0.50, deployment_tolerance=0.10)
assert ok["verdict"] == "OK"
bad = self._summ(deployment_ratio=0.38, design_fraction=0.50, deployment_tolerance=0.10)
assert bad["verdict"] == "ATTENTION"


class TestSummarizeDeployment:
"""집계 배치율 미달 판정(순수 함수)."""
Expand Down
Loading