kjh2064
da0e1b0f7e
매크로·실적·펀더멘털·공매도수급·호가미시구조·대내외 변수 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 일일 스케줄(장마감 후) + 파이프라인 계약 검증 게이트
57 lines
2.4 KiB
Python
57 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _read(path: Path) -> str:
|
|
return path.read_text(encoding="utf-8", errors="replace") if path.exists() else ""
|
|
|
|
|
|
def main() -> int:
|
|
files = {
|
|
"workflow": ROOT / ".gitea" / "workflows" / "qualitative_sell_strategy.yml",
|
|
"build_inputs": ROOT / "tools" / "build_qualitative_sell_inputs_v1.py",
|
|
"build_satellite": ROOT / "tools" / "build_satellite_candidate_recommendations_v1.py",
|
|
"evaluate": ROOT / "tools" / "evaluate_qualitative_sell_strategy_accuracy_v1.py",
|
|
"store": ROOT / "src" / "quant_engine" / "qualitative_sell_strategy_store_v1.py",
|
|
"package": ROOT / "package.json",
|
|
}
|
|
errors: list[str] = []
|
|
|
|
for name, path in files.items():
|
|
if not path.exists():
|
|
errors.append(f"missing:{name}")
|
|
|
|
checks = {
|
|
"build_inputs_flags": ("--store-backend" in _read(files["build_inputs"]) and "--store-location" in _read(files["build_inputs"])),
|
|
"build_satellite_flags": ("--store-backend" in _read(files["build_satellite"]) and "--store-location" in _read(files["build_satellite"])),
|
|
"evaluate_flags": ("--store-backend" in _read(files["evaluate"]) and "--store-location" in _read(files["evaluate"])),
|
|
"store_contract": ("resolve_store_path" in _read(files["store"]) and "QualitativeSellStoreSpec" in _read(files["store"])),
|
|
"workflow_mentions_mock_validation": ("validate_kis_api_credentials_v1.py" in _read(files["workflow"])),
|
|
"workflow_has_schedule": ("schedule:" in _read(files["workflow"]) and "workflow_dispatch:" in _read(files["workflow"])),
|
|
"package_scripts": ("ops:sell-build" in _read(files["package"]) and "ops:sell-eval" in _read(files["package"]) and "ops:sell-validate" in _read(files["package"])),
|
|
}
|
|
|
|
for key, ok in checks.items():
|
|
if not ok:
|
|
errors.append(key)
|
|
|
|
result = {
|
|
"formula_id": "QUALITATIVE_SELL_STRATEGY_PIPELINE_V1",
|
|
"gate": "PASS" if not errors else "FAIL",
|
|
"checks": checks,
|
|
"errors": errors,
|
|
}
|
|
out = ROOT / "Temp" / "qualitative_sell_strategy_pipeline_v1.json"
|
|
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if not errors else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|