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 일일 스케줄(장마감 후) + 파이프라인 계약 검증 게이트
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.quant_engine.qualitative_sell_strategy_store_v1 import (
|
|
QualitativeSellStoreSpec,
|
|
fetch_recent_sell_strategy_results,
|
|
insert_satellite_recommendation,
|
|
insert_sell_strategy_result,
|
|
resolve_store_path,
|
|
)
|
|
|
|
|
|
def test_insert_and_fetch_sell_strategy_result(tmp_path):
|
|
db_path = tmp_path / "test.db"
|
|
result = {
|
|
"code": "005930",
|
|
"generated_at": "2026-06-21T12:00:00+09:00",
|
|
"decision": {
|
|
"action": "TRIM_REVIEW_PARTIAL",
|
|
"conviction": "MEDIUM",
|
|
"market_regime": "TECHNICAL_MARKET",
|
|
"composite_score": 0.42,
|
|
"rationale": "test rationale",
|
|
},
|
|
}
|
|
insert_sell_strategy_result(db_path, result)
|
|
rows = fetch_recent_sell_strategy_results(db_path, "005930")
|
|
assert len(rows) == 1
|
|
assert rows[0]["action"] == "TRIM_REVIEW_PARTIAL"
|
|
assert rows[0]["composite_score"] == 0.42
|
|
|
|
|
|
def test_fetch_returns_empty_list_when_db_missing(tmp_path):
|
|
rows = fetch_recent_sell_strategy_results(tmp_path / "nonexistent.db", "005930")
|
|
assert rows == []
|
|
|
|
|
|
def test_multiple_inserts_ordered_by_generated_at_desc(tmp_path):
|
|
db_path = tmp_path / "test.db"
|
|
for ts in ("2026-06-19T12:00:00", "2026-06-21T12:00:00", "2026-06-20T12:00:00"):
|
|
insert_sell_strategy_result(db_path, {
|
|
"code": "005930", "generated_at": ts,
|
|
"decision": {"action": "HOLD_NO_CONFLUENCE"},
|
|
})
|
|
rows = fetch_recent_sell_strategy_results(db_path, "005930")
|
|
assert [r["generated_at"] for r in rows] == ["2026-06-21T12:00:00", "2026-06-20T12:00:00", "2026-06-19T12:00:00"]
|
|
|
|
|
|
def test_insert_satellite_recommendation(tmp_path):
|
|
db_path = tmp_path / "test.db"
|
|
insert_satellite_recommendation(db_path, "2026-06-21T12:00:00+09:00", {
|
|
"ticker": "042700",
|
|
"score": {"satellite_action": "BUY_CANDIDATE", "attractiveness_score": 0.6, "market_regime": "PERFORMANCE_MARKET"},
|
|
})
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_path)
|
|
row = conn.execute("SELECT ticker, satellite_action, attractiveness_score FROM satellite_recommendations").fetchone()
|
|
conn.close()
|
|
assert row == ("042700", "BUY_CANDIDATE", 0.6)
|
|
|
|
|
|
def test_resolve_store_path_supports_sqlite(tmp_path):
|
|
db_path = resolve_store_path(QualitativeSellStoreSpec(location=tmp_path / "qualitative.db"), ROOT)
|
|
assert str(db_path).endswith("qualitative.db")
|