kjh2064
ee3e799de1
주요 변경: - tools/build_rebalance_engine_v1.py: REBALANCE_ENGINE_V1 신규 * account_snapshot 직접 합산(_build_snap_position_map) → 소수주 분리 행 병합 * 레짐 소스 macro.REGIME_PRELIM 최우선 (GAS 와 동일) - src/gas_adapter_parts/gdf_06_rebalance.gs: runRebalanceSheet_() 신규 * Logger.log / getSpreadsheet_() 로 run_all 연동 수정 - src/gas_adapter_parts/gdc_01_fetch_fundamentals.gs * _mergePositionRecord_(): 소수주 중복 행 합산 신규 * parseInt → parseFloat (qty, availQty) - src/gas_adapter_parts/gdf_01_price_metrics.gs * 미보유 종목 SELL_READY → WATCH_EXIT_SIGNAL - spec/41_release_dag.yaml: build_rebalance_sheet 노드 추가 (step_count 63) - spec/51_formula_lifecycle_registry.yaml: REBALANCE_ENGINE_V1 등록 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
92 lines
4.6 KiB
Python
92 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
import sys
|
|
import json
|
|
import yaml
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.quant_engine.exit_decisions import (
|
|
compute_stop_price_core,
|
|
compute_stop_action_ladder,
|
|
compute_dynamic_heat_thresholds,
|
|
)
|
|
from src.quant_engine.compute_formula_outputs import (
|
|
compute_imputed_data_exposure,
|
|
compute_cash_recovery_optimizer,
|
|
krx_tick_unit,
|
|
)
|
|
|
|
def test_cash_shortfall_monotonicity():
|
|
# 1. Cash Shortfall Monotonicity: 현금 부족액이 증가하면 매도 계획의 예상 회수액과 대상 종목 수는 단조 증가해야 함.
|
|
sell_candidates = [
|
|
{"Ticker": "005930", "Name": "삼성전자", "Sell_Qty": 100, "Sell_Limit_Price": 70000, "Cash_Preserve_Ratio": 100, "Cash_Preserve_Style": "FULL"},
|
|
{"Ticker": "000660", "Name": "SK하이닉스", "Sell_Qty": 50, "Sell_Limit_Price": 180000, "Cash_Preserve_Ratio": 100, "Cash_Preserve_Style": "FULL"},
|
|
]
|
|
|
|
res_small = compute_cash_recovery_optimizer(sell_candidates, 1_000_000)
|
|
res_large = compute_cash_recovery_optimizer(sell_candidates, 10_000_000)
|
|
|
|
seq_small = res_small["cash_recovery_plan_json"]["sell_sequence"]
|
|
seq_large = res_large["cash_recovery_plan_json"]["sell_sequence"]
|
|
|
|
assert len(seq_large) >= len(seq_small), "Item count should not decrease when shortfall increases"
|
|
assert res_large["cash_recovery_plan_json"]["expected_total_krw"] >= res_small["cash_recovery_plan_json"]["expected_total_krw"], "Expected recovered cash should not decrease when shortfall increases"
|
|
print("[PASS] INV_CASH_SHORTFALL_MONOTONICITY")
|
|
|
|
def test_market_risk_monotonicity():
|
|
# 2. Market Risk Monotonicity: regime이 RISK_OFF 일 때 max position count 등 제약이 강화되는지 검증
|
|
# GatherTradingData.json 의 settings 구조 확인
|
|
json_path = ROOT / "GatherTradingData.json"
|
|
if json_path.exists():
|
|
raw = json.loads(json_path.read_text(encoding="utf-8"))
|
|
settings = raw.get("data", {}).get("settings", {})
|
|
pos_normal = settings.get("position_count_max_normal", 12)
|
|
pos_risk_off = settings.get("position_count_max_risk_off", 8)
|
|
assert pos_risk_off < pos_normal, "Risk off limit must be strictly more conservative than normal limit"
|
|
print("[PASS] INV_MARKET_RISK_MONOTONICITY")
|
|
|
|
def test_missing_data_confidence():
|
|
# 3. Missing Data Confidence: domain coverage가 낮아지면 weighted coverage가 하락하고 imputed field ratio(ifr)가 상승하여 confidence가 낮아져야 함.
|
|
coverage_high = {"fundamental_core": 1.0, "realized_outcome": 1.0, "trade_quality": 1.0, "pattern": 1.0, "alpha_eval": 1.0}
|
|
coverage_low = {"fundamental_core": 0.5, "realized_outcome": 0.5, "trade_quality": 0.5, "pattern": 0.5, "alpha_eval": 0.5}
|
|
|
|
res_high = compute_imputed_data_exposure(coverage_high, 100.0)
|
|
res_low = compute_imputed_data_exposure(coverage_low, 100.0)
|
|
|
|
assert res_low["weighted_coverage"] < res_high["weighted_coverage"], "Weighted coverage must drop"
|
|
assert res_low["imputed_field_ratio"] > res_high["imputed_field_ratio"], "Imputed field ratio must rise"
|
|
assert res_low["effective_confidence_honest"] < res_high["effective_confidence_honest"], "Confidence must drop"
|
|
print("[PASS] INV_MISSING_DATA_CONFIDENCE")
|
|
|
|
def test_stale_price_zero_quantity():
|
|
# 4. Stale Price / Data Missing: 필수 데이터 결측 시 stop price 계산이 PASS가 되지 못하고 DATA_MISSING 경고가 되며 fallback 로직이 작동하는지 검증
|
|
res_missing = compute_stop_price_core(entry_price=10000.0, atr20=None, current_price=10000.0)
|
|
assert res_missing["stop_price_status"].startswith("DATA_MISSING"), "Missing ATR must trigger DATA_MISSING"
|
|
assert res_missing["stop_price"] == 10000.0 * 0.92, "Fallback stop price must be 92% of entry price"
|
|
print("[PASS] INV_STALE_PRICE_ZERO_QUANTITY")
|
|
|
|
def main():
|
|
try:
|
|
test_cash_shortfall_monotonicity()
|
|
test_market_risk_monotonicity()
|
|
test_missing_data_confidence()
|
|
test_stale_price_zero_quantity()
|
|
except AssertionError as e:
|
|
print(f"[FAIL] Invariant check failed: {e}")
|
|
sys.exit(1)
|
|
|
|
result_path = ROOT / "Temp" / "property_test_result_v1.json"
|
|
result_path.parent.mkdir(parents=True, exist_ok=True)
|
|
result_path.write_text(json.dumps({
|
|
"status": "PASS",
|
|
"tests_run": 4,
|
|
"timestamp": "2026-06-07T15:00:00Z"
|
|
}, indent=2), encoding="utf-8")
|
|
print(f"Property tests completed successfully. Results saved to {result_path}")
|
|
sys.exit(0)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|