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
8 changes: 5 additions & 3 deletions config/baskets.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -100,9 +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로 완화 — 이걸 넘으면 절사가 아니라 진짜 이상이다.
# 소액 구조적 절사: ETF 1주(약 12.8만) 단위라 배치율 미달이 상시 존재하고, 적립
# 직후에는 '부족분 < 1주'로 매수가 보류돼 미달이 최대 1주가/총액(잔고 40만 시점
# 32%p)까지 벌어진다(docs/POCKET_TRACK_PLAN.md §3 — 정상, 적립 누적 시 수렴).
# 이 값은 '하한(floor)'이고, 헬스가 구조 하한(1주가/총액)과의 max로 자동 완화·
# 조임한다 — 잔고가 커지면 이 값이 다시 지배한다.
monitoring:
deployment_tolerance: 0.10
rebalance:
Expand Down
25 changes: 25 additions & 0 deletions core/operator_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,31 @@ def summarize_blockers(blockers: dict[str, Any] | None) -> dict[str, Any]:
}


def structural_deployment_tolerance(
max_share_price: float | None,
total_value: float | None,
floor_tolerance: float = 0.05,
) -> float:
"""배치율 허용 오차의 구조 하한 — 정수 주식 절사가 만드는 불가피한 미달을 반영(순수).

적립 직후에는 '부족분 < 1주 가격'인 동안 매수가 보류되므로(설계), 미달폭이 최대
1주 가격/총액까지 벌어진다(예: 잔고 40만·ETF 1주 12.8만 → 32%p). 이 구간을 고정
허용치로 재면 정상 적립 과정이 매일 ATTENTION을 울린다(경보 피로). 반대로 잔고가
커지면 구조 하한이 저절로 조여져(200만 → 6.4%p) floor가 다시 지배한다 —
진짜 이상(예: 매수 자체 불능 → 미달 50%p)은 어느 시점에도 잡힌다.

반환: max(floor_tolerance, max_share_price/total_value). 입력 불충분 시 floor.
"""
try:
price = float(max_share_price or 0)
total = float(total_value or 0)
except (TypeError, ValueError):
return float(floor_tolerance)
if price <= 0 or total <= 0:
return float(floor_tolerance)
return max(float(floor_tolerance), price / total)


def summarize_deployment(
deployment_ratio: float | None,
design_fraction: float | None,
Expand Down
22 changes: 17 additions & 5 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1722,21 +1722,33 @@ def _design_fraction(cfg: dict) -> float:
.first()
)
last_dates.append(snap.date if snap else None)
position_count += len(get_all_positions(account_key=key) or [])
positions_b = get_all_positions(account_key=key) or []
position_count += len(positions_b)
# 배치율은 부가 신호 — 계산 실패(예: baskets.yaml에 float 불가한
# target_stock_weight 오타)가 핵심 신호인 결측/staleness 감지를
# 통째로 삼키지 않도록 자체 try로 격리한다.
# 바스켓별 허용 오차(monitoring.deployment_tolerance, 기본 5%p)를
# 반영해 '허용 초과분'이 가장 큰 바스켓을 고른다 — 원시 미달폭
# 최대로 고르면 관찰용 트랙(허용 완화)이 진짜 감시 대상을 가린다.
# 허용 오차 = max(바스켓 설정값, 구조 하한 1주가/총액) — 적립 직후
# '부족분 < 1주'로 매수가 보류되는 동안의 불가피한 미달(예: 잔고
# 40만·ETF 12.8만 → 32%p)이 정상인데 매일 울리는 것을 방지. 잔고가
# 크면 구조 하한이 저절로 조여져 설정값이 다시 지배한다.
# '허용 초과분'이 가장 큰 바스켓을 고른다 — 원시 미달폭 최대로
# 고르면 관찰용 트랙(허용 완화)이 진짜 감시 대상을 가린다.
try:
if snap and snap.total_value and snap.total_value > 0:
from core.operator_health import structural_deployment_tolerance

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(cfg_b)
tol_b = float(
floor_tol = float(
(cfg_b.get("monitoring") or {}).get("deployment_tolerance", 0.05)
)
max_price = max(
(float(p.avg_price or 0) for p in positions_b), default=0.0,
)
tol_b = structural_deployment_tolerance(
max_price, float(snap.total_value), floor_tol,
)
violation = (design - dep_ratio) - tol_b
if worst_shortfall is None or violation > worst_shortfall:
worst_shortfall = violation
Expand Down
35 changes: 35 additions & 0 deletions tests/test_operator_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from types import SimpleNamespace

import pytest

from core.operator_health import (
summarize_runtime_state,
summarize_blockers,
Expand Down Expand Up @@ -225,6 +227,39 @@ def test_per_basket_tolerance_relaxes_attention(self):
assert bad["verdict"] == "ATTENTION"


class TestStructuralDeploymentTolerance:
"""적립 직후 구조적 미달(부족분 < 1주 → 매수 보류)의 허용 하한 — 8/1 입금 시나리오."""

def _f(self, price, total, floor=0.10):
from core.operator_health import structural_deployment_tolerance
return structural_deployment_tolerance(price, total, floor)

def test_month2_deposit_scenario_not_alarmed(self):
# 잔고 40만(30만+입금 10만), KODEX 1주 12.8만 보유: 부족분 7.2만 < 1주라
# 매수 보류가 설계 — 미달 18%p는 구조 하한 32%p 이내여야 한다(무경보).
from core.operator_health import summarize_deployment
tol = self._f(128_000, 400_000) # = 0.32
assert tol == pytest.approx(0.32)
out = summarize_deployment(0.32, 0.50, tolerance=tol) # 미달 18%p
assert out["verdict"] == "OK"

def test_genuine_failure_still_caught(self):
# 매수 자체가 안 된 이상 상태(주식 0): 미달 50%p > 구조 하한 32%p → ATTENTION
from core.operator_health import summarize_deployment
tol = self._f(128_000, 400_000)
out = summarize_deployment(0.0, 0.50, tolerance=tol)
assert out["verdict"] == "ATTENTION"

def test_tightens_as_balance_grows(self):
# 잔고 200만이면 구조 하한 6.4%p < floor 10%p → floor가 지배(자동 조임)
assert self._f(128_000, 2_000_000) == pytest.approx(0.10)

def test_floor_when_inputs_missing(self):
assert self._f(None, 400_000) == pytest.approx(0.10)
assert self._f(128_000, 0) == pytest.approx(0.10)
assert self._f("x", 400_000) == pytest.approx(0.10)


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

Expand Down
Loading