비기계적 매도전략(가치보존) + 위성종목 추천 엔진 추가
매크로·실적·펀더멘털·공매도수급·호가미시구조·대내외 변수 5개 독립 팩터군의 confluence(최소 3/5 합의) 없이는 매도 트리거를 금지하는 정성적 매도판단 엔진과, 보유종목 제외 위성후보 추천 로직을 추가한다. - 단일 팩터 임계값 돌파만으로는 매도 신호를 생성하지 않음 (mechanical_sell_prohibited=true) - 데이터 결측 시 항상 DATA_MISSING/INSUFFICIENT_DATA_NO_ACTION — 추정값으로 채우지 않음 - KIS 호가10단계·공매도거래비중 + Naver 시세/수급 스크래핑 입력 연동 - SQLite 시계열 저장 + 사후 적중률 자체평가 (evaluate_qualitative_sell_strategy_accuracy_v1) - Gitea 일일 스케줄(장마감 후) + 파이프라인 계약 검증 게이트
This commit is contained in:
@@ -0,0 +1,143 @@
|
||||
"""qualitative_sell_strategy_v1 자체 평가 루프 — "한 번 만들고 끝"이 아니라 결정이
|
||||
실제로 가치를 보존했는지 사후 검증한다(30년 시니어 퀀트의 핵심 습관: 판단 → 결과 →
|
||||
재보정). 기존 T+5/T+20 outcome ledger(proposal_evaluation_history)와 별개로,
|
||||
qualitative_sell_strategy_store_v1.db에 쌓인 SQLite 시계열을 사용한다 — GAS/xlsx와
|
||||
무관하므로 이 모듈만의 독립 평가 루프를 구성해도 기존 시스템과 충돌하지 않는다.
|
||||
|
||||
판정 기준(가치보존 관점, 기계적 승률 게임이 아님):
|
||||
- EXIT_REVIEW_FULL / TRIM_REVIEW_PARTIAL(매도방향) → 이후 가격이 하락했으면
|
||||
"가치보존 성공"(매도가 손실을 막았다). 상승했으면 "기회비용 발생"(조급한 매도).
|
||||
- HOLD_ADD_CONVICTION(지지방향) → 이후 가격이 상승했으면 성공.
|
||||
- HOLD_NO_CONFLUENCE / INSUFFICIENT_DATA_NO_ACTION → 방향성 주장이 없으므로 평가 대상 제외.
|
||||
|
||||
표본이 부족하면(DATA_GATED) 추정하지 않고 명시적으로 보류한다 — honest_proof_score와
|
||||
동일한 원칙(spec/algorithm_guidance_proof 계열).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from src.quant_engine.qualitative_sell_strategy_store_v1 import QualitativeSellStoreSpec, resolve_store_path
|
||||
|
||||
MIN_HOLDING_DAYS = 5 # T+5 수준 — 너무 짧으면 노이즈, 너무 길면 표본 희소
|
||||
MIN_SAMPLE_FOR_HIT_RATE = 10 # 이보다 적으면 hit_rate를 신뢰 구간 없이 표기하지 않음(DATA_GATED)
|
||||
|
||||
|
||||
def _scoreable_direction(action: str) -> int | None:
|
||||
if action in {"EXIT_REVIEW_FULL", "TRIM_REVIEW_PARTIAL"}:
|
||||
return -1 # 매도 방향 — 가격 하락이 "성공"
|
||||
if action == "HOLD_ADD_CONVICTION":
|
||||
return 1 # 지지 방향 — 가격 상승이 "성공"
|
||||
return None # HOLD_NO_CONFLUENCE / INSUFFICIENT_DATA_NO_ACTION — 평가 제외
|
||||
|
||||
|
||||
def load_scoreable_decisions(db_path: Path, min_age_days: int = MIN_HOLDING_DAYS) -> list[dict[str, Any]]:
|
||||
if not db_path.exists():
|
||||
return []
|
||||
cutoff = (dt.date.today() - dt.timedelta(days=min_age_days)).isoformat()
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"SELECT code, generated_at, action, conviction, market_regime, composite_score "
|
||||
"FROM sell_strategy_results WHERE generated_at <= ? ORDER BY generated_at",
|
||||
(cutoff,),
|
||||
).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def evaluate_decision(decision: dict[str, Any], price_at_decision: float, price_after: float) -> dict[str, Any] | None:
|
||||
direction = _scoreable_direction(decision["action"])
|
||||
if direction is None or not price_at_decision or price_at_decision <= 0:
|
||||
return None
|
||||
realized_return_pct = (price_after / price_at_decision - 1.0) * 100.0
|
||||
success = (direction * realized_return_pct) > 0 # 방향 일치 시 성공
|
||||
return {
|
||||
**decision,
|
||||
"price_at_decision": price_at_decision,
|
||||
"price_after": price_after,
|
||||
"realized_return_pct": round(realized_return_pct, 4),
|
||||
"success": success,
|
||||
}
|
||||
|
||||
|
||||
def build_accuracy_report(db_path: Path, price_lookup: dict[str, dict[str, float]]) -> dict[str, Any]:
|
||||
"""price_lookup: {code: {generated_at_date_iso: close_price}} — 호출측이 실제 가격
|
||||
히스토리(fetch_naver_market_data_v1 등)로 조립해 주입한다. 이 함수는 가격을 추정하지
|
||||
않는다 — 주어진 값만 사용."""
|
||||
decisions = load_scoreable_decisions(db_path)
|
||||
evaluated: list[dict[str, Any]] = []
|
||||
skipped_no_price = 0
|
||||
for decision in decisions:
|
||||
prices = price_lookup.get(decision["code"], {})
|
||||
decision_date = decision["generated_at"][:10]
|
||||
price_at = prices.get(decision_date)
|
||||
future_date = (dt.date.fromisoformat(decision_date) + dt.timedelta(days=MIN_HOLDING_DAYS)).isoformat()
|
||||
price_after = prices.get(future_date)
|
||||
if price_at is None or price_after is None:
|
||||
skipped_no_price += 1
|
||||
continue
|
||||
result = evaluate_decision(decision, price_at, price_after)
|
||||
if result is not None:
|
||||
evaluated.append(result)
|
||||
|
||||
scored = [e for e in evaluated if e is not None]
|
||||
if len(scored) < MIN_SAMPLE_FOR_HIT_RATE:
|
||||
return {
|
||||
"status": "DATA_GATED",
|
||||
"scored_sample_count": len(scored),
|
||||
"min_sample_required": MIN_SAMPLE_FOR_HIT_RATE,
|
||||
"note": "표본 부족 — hit_rate를 산출하지 않음(추정 금지). 결정 누적과 가격 매칭이 더 필요.",
|
||||
"skipped_no_price": skipped_no_price,
|
||||
}
|
||||
|
||||
hit_rate_pct = round(100.0 * sum(1 for e in scored if e["success"]) / len(scored), 2)
|
||||
return {
|
||||
"status": "OK",
|
||||
"scored_sample_count": len(scored),
|
||||
"hit_rate_pct": hit_rate_pct,
|
||||
"evaluations": scored,
|
||||
"skipped_no_price": skipped_no_price,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("--sqlite-db", type=Path,
|
||||
default=ROOT / "outputs" / "qualitative_sell_strategy" / "qualitative_sell_strategy.db")
|
||||
ap.add_argument("--store-backend", default="sqlite", help="Storage backend contract placeholder (sqlite today, postgresql planned)")
|
||||
ap.add_argument("--store-location", default=None, help="Backend location/DSN. sqlite path or future postgres DSN.")
|
||||
ap.add_argument("--price-lookup-json", type=Path, default=None,
|
||||
help='{"code": {"YYYY-MM-DD": close_price, ...}} 형식 — 미지정 시 가격 매칭 없이 표본 카운트만 보고')
|
||||
args = ap.parse_args()
|
||||
db_path = resolve_store_path(
|
||||
QualitativeSellStoreSpec(
|
||||
backend=args.store_backend,
|
||||
location=args.store_location or args.sqlite_db,
|
||||
),
|
||||
ROOT,
|
||||
)
|
||||
|
||||
price_lookup: dict[str, dict[str, float]] = {}
|
||||
if args.price_lookup_json and args.price_lookup_json.exists():
|
||||
price_lookup = json.loads(args.price_lookup_json.read_text(encoding="utf-8"))
|
||||
|
||||
report = build_accuracy_report(db_path, price_lookup)
|
||||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user