kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

섹터 리포트와 대표종목 모니터 고도화

Browse Source
This commit is contained in:
김재현
2026-06-15 02:30:02 +09:00
parent 82ca4ddbfd
commit 2439980730
5 changed files with 192 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/quant_engine/etf_representative_monitor.py
+48 -13
View File
@@ -13,6 +13,29 @@ ETF_NAME_HINTS = (
"SOL", "TIMEFOLIO", "WOORI", "PLUS", "NPLUS", "TREX", "FOCUS", "KIWOOM",
)
ROBOTICS_FALLBACK_PROXY = {
"Sector": "로보틱스",
"Proxy_Ticker": "0190C0",
"Proxy_Name": "RISE 현대차고정피지컬AI",
"Proxy_Type": "ETF",
"Sector_Rank": 12,
"SmartMoney_5D_KRW": 0.0,
"Sector_Ret20D": 0.0,
}
ROBOTICS_FALLBACK_UNIVERSE = [
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "005380", "Constituent_Name": "현대차", "Weight": 0.2402, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "012330", "Constituent_Name": "현대모비스", "Weight": 0.1588, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "011070", "Constituent_Name": "LG이노텍", "Weight": 0.1450, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "000270", "Constituent_Name": "기아", "Weight": 0.1234, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "307950", "Constituent_Name": "현대오토에버", "Weight": 0.0899, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "277810", "Constituent_Name": "레인보우로보틱스", "Weight": 0.0673, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "064400", "Constituent_Name": "LG씨엔에스", "Weight": 0.0519, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "454910", "Constituent_Name": "두산로보틱스", "Weight": 0.0367, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "108490", "Constituent_Name": "로보티즈", "Weight": 0.0240, "Is_ETF": False},
{"Sector": "로보틱스", "Proxy_Ticker": "0190C0", "Proxy_Name": "RISE 현대차고정피지컬AI", "Proxy_Type": "ETF", "Constituent_Code": "058610", "Constituent_Name": "에스피지", "Weight": 0.0173, "Is_ETF": False},
]
def _parse_jsonish(value: Any) -> Any:
if isinstance(value, (dict, list)):
@@ -174,6 +197,8 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
continue
if _txt(row.get("Proxy_Type")).upper() == "ETF":
etf_sectors[sector] = row
if "로보틱스" not in etf_sectors:
etf_sectors["로보틱스"] = ROBOTICS_FALLBACK_PROXY
sector_candidates: dict[str, list[dict[str, Any]]] = defaultdict(list)
core_by_ticker: dict[str, dict[str, Any]] = {}
@@ -201,9 +226,12 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
if _txt(row.get("Status"), "OK").upper() not in {"OK", "ACTIVE", "LIVE"}:
continue
universe_candidates[sector].append(row)
if "로보틱스" not in universe_candidates:
universe_candidates["로보틱스"] = ROBOTICS_FALLBACK_UNIVERSE.copy()
rows: list[dict[str, Any]] = []
for sector, proxy in sorted(etf_sectors.items(), key=lambda item: (_num(item[1].get("Sector_Rank"), 999), -abs(_num(item[1].get("SmartMoney_5D_KRW"), 0.0)))):
target_rep_count = 5 if sector == "로보틱스" else 3
fallback_rows = sorted(
sector_candidates.get(sector, []),
key=lambda r: (
@@ -213,31 +241,36 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
-_num(r.get("Ret10D"), 0.0),
),
)
# ETF 대표주는 구성비 내림차순을 1차 기준으로 고정한다.
# live score는 동일 비중/동일 구성일 때만 보조 판단으로 사용한다.
universe_rows = sorted(
universe_candidates.get(sector, []),
key=lambda r: _constituent_priority_score(
r,
core_by_ticker.get(_txt(r.get("Constituent_Code")))
or next((x for x in fallback_rows if _txt(x.get("Ticker")) == _txt(r.get("Constituent_Code"))), None),
key=lambda r: (
-_num(r.get("Weight"), 0.0),
_constituent_priority_score(
r,
core_by_ticker.get(_txt(r.get("Constituent_Code")))
or next((x for x in fallback_rows if _txt(x.get("Ticker")) == _txt(r.get("Constituent_Code"))), None),
),
),
)
basket_items: list[dict[str, Any]] = []
selected_specs: list[tuple[str, dict[str, Any]]] = [("ETF_CONSTITUENT_WEIGHT", row) for row in universe_rows[:3]]
selected_tickers = {_txt(row.get("Constituent_Code")) for row in universe_rows[:3]}
if len(selected_specs) < 3:
selected_specs: list[tuple[str, dict[str, Any]]] = [("ETF_CONSTITUENT_WEIGHT", row) for row in universe_rows[:target_rep_count]]
selected_tickers = {_txt(row.get("Constituent_Code")) for row in universe_rows[:target_rep_count]}
if len(selected_specs) < target_rep_count:
for row in fallback_rows:
ticker = _txt(row.get("Ticker"))
if not ticker or ticker in selected_tickers:
continue
selected_specs.append(("SECTOR_LIQUIDITY_FALLBACK", row))
selected_tickers.add(ticker)
if len(selected_specs) >= 3:
if len(selected_specs) >= target_rep_count:
break
if not selected_specs:
selected_specs = [("SECTOR_LIQUIDITY_FALLBACK", row) for row in fallback_rows[:3]]
selected_specs = [("SECTOR_LIQUIDITY_FALLBACK", row) for row in fallback_rows[:target_rep_count]]
rep_source = "ETF_CONSTITUENT_WEIGHT" if universe_rows else "SECTOR_LIQUIDITY_FALLBACK"
rep_basis_detail = "ETF_WEIGHT_PRIMARY"
if universe_rows and len(universe_rows) < 3 and len(selected_specs) >= 3:
if universe_rows and len(universe_rows) < target_rep_count and len(selected_specs) >= target_rep_count:
rep_basis_detail = "ETF_WEIGHT_PRIMARY_PLUS_SECTOR_TOPUP"
if not universe_rows:
rep_basis_detail = "SECTOR_LIQUIDITY_FALLBACK"
@@ -283,7 +316,7 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
_txt(spec.get("Constituent_Code")),
_txt(spec.get("Constituent_Name")),
))
if len(basket_items) < 3:
if len(basket_items) < target_rep_count:
used_tickers = {item["ticker"] for item in basket_items}
for rep in fallback_rows:
ticker = _txt(rep.get("Ticker"))
@@ -291,7 +324,7 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
continue
basket_items.append(_build_rep_item(rep, {"Weight": ""}, proxy, "SECTOR_LIQUIDITY_FALLBACK"))
used_tickers.add(ticker)
if len(basket_items) >= 3:
if len(basket_items) >= target_rep_count:
break
if not basket_items:
continue
@@ -313,6 +346,7 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
"etf_proxy_ticker": _txt(proxy.get("Proxy_Ticker")),
"etf_proxy_name": _txt(proxy.get("Proxy_Name")),
"etf_proxy_type": _txt(proxy.get("Proxy_Type")),
"universe_source": _txt(proxy.get("Universe_Source"), "DEFAULT_TEMPLATE"),
"sector_rank": proxy.get("Sector_Rank", ""),
"sector_score": proxy.get("Sector_Score", ""),
"sector_smart_money_5d_krw": proxy.get("SmartMoney_5D_KRW", ""),
@@ -348,7 +382,7 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
"basket_quality_state": basket_quality_state,
"representatives": basket_items,
"monitor_reason": (
"ETF 구성비중 상위 3종목이 같은 방향으로 정렬"
f"ETF 구성비중 상위 {target_rep_count}종목이 같은 방향으로 정렬"
if basket_state == "BUY_REVIEW"
else "대표 종목 바스켓 추세 확인 중" if basket_state == "TRACK"
else "유동성/추세 보수 모니터링"
@@ -390,6 +424,7 @@ def build_etf_representative_monitor(payload: dict[str, Any]) -> dict[str, Any]:
"sector_flow_rows": len(sector_flow),
"core_satellite_rows": len(core_satellite),
"sector_universe_rows": len(sector_universe),
"template_source_count": sum(1 for r in rows if str(r.get("universe_source") or "").upper() == "DEFAULT_TEMPLATE"),
},
}
return result