12b34929d7
- new energy_irpc collector parsing IRPC performance-highlights table (net profit/EBITDA/ROE margins, latest period 3M26: +10.27%) - factor energy_irpc_net_margin (sign +1) wired into refining_energy/ exploration/utilities, extending the energy theme beyond TOP - scheduler job + dashboard fetch + sources table row (now 9 sources) - tests: parse (incl paren-negatives), value-key resolution, direction; suite 368 OK. Independent review passed: true - Phase B feasibility: REIC/EPPO/NBTC/PTTEP are JS-rendered or anti-bot (recorded deferred in plan); IRPC was the clean server-rendered win
428 lines
24 KiB
Python
428 lines
24 KiB
Python
"""Real multi-theme dashboard assembly.
|
||
|
||
Replaces the tourism-only fixture dashboard with a REAL, multi-theme view of the
|
||
Thai economy and the SET50 universe. For each of the 3 themes it reads the live
|
||
Thai factor data (via the daily cache), computes a uniform z-score surprise, and
|
||
assembles:
|
||
|
||
- themes: list of {id, label_th, frequency, surprise, read: {...}, thesis}
|
||
- macro: Thai macro backdrop (consumption/inflation/unemployment/tourism)
|
||
- board: per-symbol combined score (60/40) that the simulation also uses
|
||
- sources: provenance table (from -> source -> as_of -> fetched_at)
|
||
|
||
Only real data is used; if a required collector fails the assembly fails loudly
|
||
(no fixture fallback) per the user's explicit decision.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import statistics
|
||
from typing import Any, Callable, Optional
|
||
|
||
from . import macro_thai, themes as themes_mod
|
||
|
||
|
||
class DashboardError(Exception):
|
||
"""Raised when real data cannot be assembled (no fixture fallback)."""
|
||
|
||
|
||
def _fetch_with_cache(
|
||
cache: Any,
|
||
key: str,
|
||
fetcher: Callable[[], dict],
|
||
label: str,
|
||
) -> dict:
|
||
try:
|
||
val = cache.fetch_or_stale(key, fetcher)
|
||
if isinstance(val, dict) and "data" in val:
|
||
return val["data"]
|
||
return val or {}
|
||
except Exception as exc:
|
||
raise DashboardError(f"no real data for {label}: {exc}") from exc
|
||
|
||
|
||
def _zscore(value: float, mean: float, stdev: float) -> float:
|
||
return (value - mean) / stdev if stdev else 0.0
|
||
|
||
|
||
def _auto_read(auto_d: dict, npl_d: dict, cache: Any) -> dict:
|
||
# multi-source: volume (YoY) + credit quality (NPL)
|
||
read = {
|
||
"source": "TradingEconomics + BOT",
|
||
"frequency": "monthly",
|
||
"new_car_sales_yoy": auto_d.get("new_car_sales_yoy"),
|
||
"total_vehicle_sales": auto_d.get("total_vehicle_sales"),
|
||
"vehicle_production": auto_d.get("vehicle_production"),
|
||
"auto_exports": auto_d.get("auto_exports"),
|
||
"passenger_car_sales": auto_d.get("passenger_car_sales"),
|
||
"auto_npl_pct": npl_d.get("pct_of_npls"),
|
||
"auto_npl_amount": npl_d.get("npl_amount"),
|
||
}
|
||
read["thesis"] = (
|
||
"ยอดขายรถยนต์และสินเชื่อที่เกี่ยวข้อง (NPL) สะท้อนกำลังซื้อรถในประเทศ."
|
||
)
|
||
return read
|
||
|
||
|
||
class RealDashboard:
|
||
"""Assemble the real multi-theme dashboard from live collectors."""
|
||
|
||
def __init__(self, tourism_signals: list[dict], cache: Any,
|
||
factor_view: Optional[dict] = None):
|
||
self.tourism_signals = tourism_signals or []
|
||
self.cache = cache
|
||
self.factor_view = factor_view or {"factors": []}
|
||
|
||
def build(self) -> dict:
|
||
# 1) live theme data (real, no fallback)
|
||
from . import (auto_credit, auto_npl, bank_npl, energy_irpc, energy_thai,
|
||
bot_tourism, macro_thai, te_thailand, thai_trade)
|
||
|
||
tourism = None
|
||
try:
|
||
tourism = self.cache.fetch_or_stale("bot_tourism", lambda: bot_tourism.BotTourismSource().fetch())
|
||
except Exception:
|
||
pass # handled below as no-data
|
||
|
||
auto_d = _fetch_with_cache(
|
||
self.cache, "auto_credit/tourism",
|
||
lambda: auto_credit.fetch_auto_credit().to_dict(), "auto_credit")
|
||
npl_d = _fetch_with_cache(
|
||
self.cache, "auto_npl", lambda: auto_npl.fetch_auto_npl().to_dict(), "auto_npl")
|
||
bnpl_d = _fetch_with_cache(
|
||
self.cache, "bank_npl", lambda: bank_npl.fetch_bank_npl().to_dict(), "bank_npl")
|
||
en_d = _fetch_with_cache(
|
||
self.cache, "energy_thai", lambda: energy_thai.fetch_energy_thai().to_dict(), "energy_thai")
|
||
irpc_d = _fetch_with_cache(
|
||
self.cache, "energy_irpc", lambda: energy_irpc.fetch_energy_irpc().to_dict(), "energy_irpc")
|
||
macro_d = _fetch_with_cache(
|
||
self.cache, "macro_thai", lambda: macro_thai.fetch_macro_thai().to_dict(), "macro_thai")
|
||
trade_d = _fetch_with_cache(
|
||
self.cache, "thai_trade", lambda: thai_trade.fetch_thai_trade().to_dict(), "thai_trade")
|
||
te_d = _fetch_with_cache(
|
||
self.cache, "te_thailand", lambda: te_thailand.fetch_te_thailand().to_dict(), "te_thailand")
|
||
|
||
# 2) per-theme surprise — registry-driven (THE single source of truth)
|
||
fetched = {
|
||
"macro_thai": macro_d, "auto_credit": auto_d,
|
||
"auto_npl": npl_d, "energy_thai": en_d, "energy_irpc": irpc_d,
|
||
"bank_npl": bnpl_d,
|
||
"thai_trade": trade_d, "te_thailand": te_d,
|
||
}
|
||
tourism_surprise = self._tourism_surprise()
|
||
surprises = themes_mod.compute_theme_surprises(
|
||
fetched, tourism_surprise=tourism_surprise)
|
||
|
||
# 3) assemble theme reads + thesis (all SET50 themes, so the board and
|
||
# per-symbol view have a surprise for every theme)
|
||
themes = [
|
||
self._mk_theme("tourism", surprises.get("tourism"), tourism),
|
||
self._mk_theme("auto_credit", surprises.get("auto_credit"),
|
||
_auto_read(auto_d, npl_d, self.cache)),
|
||
self._mk_theme("refining_energy", surprises.get("refining_energy"),
|
||
{**en_d, "irpc_net_margin_pct": (irpc_d or {}).get("net_margin_pct"),
|
||
"source": "TOP + IRPC investor"}),
|
||
]
|
||
# macro-proxy reads for the expanded SET50 themes (deterministic)
|
||
proxy_reads = {
|
||
"banks": {"source": "BOT macro + NPL + TE (rate/credit)",
|
||
"frequency": "monthly",
|
||
"invest_yoy": macro_d.get("private_investment_yoy"),
|
||
"inflation_yoy": macro_d.get("headline_inflation_yoy"),
|
||
"bank_npl_pct": (bnpl_d or {}).get("pct_of_npls"),
|
||
"interest_rate_pct": (te_d or {}).get("interest_rate_pct"),
|
||
"loan_growth": (te_d or {}).get("loans_to_fin_corp")},
|
||
"retail": {"source": "BOT macro + TE (retail/confidence)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy"),
|
||
"retail_sales_yoy": (te_d or {}).get("retail_sales_yoy"),
|
||
"consumer_confidence": (te_d or {}).get("consumer_confidence")},
|
||
"consumer_staples": {"source": "BOT macro + TE (retail/confidence)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy"),
|
||
"retail_sales_yoy": (te_d or {}).get("retail_sales_yoy"),
|
||
"consumer_confidence": (te_d or {}).get("consumer_confidence")},
|
||
"telecom_it": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"property": {"source": "BOT macro + TE (property)", "frequency": "monthly",
|
||
"invest_yoy": macro_d.get("private_investment_yoy"),
|
||
"property_prices_yoy": (te_d or {}).get("property_prices_yoy"),
|
||
"business_confidence": (te_d or {}).get("business_confidence")},
|
||
"utilities": {"source": "BOT macro (mfg)", "frequency": "monthly",
|
||
"mfg_yoy": macro_d.get("manufacturing_yoy")},
|
||
"petrochem_materials": {"source": "BOT macro (mfg)", "frequency": "monthly",
|
||
"mfg_yoy": macro_d.get("manufacturing_yoy")},
|
||
"healthcare": {"source": "BOT macro (consumption)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy")},
|
||
"nonbank_finance": {"source": "BOT macro + TE (credit/confidence)", "frequency": "monthly",
|
||
"consumption_yoy": macro_d.get("private_consumption_yoy"),
|
||
"consumer_credit": (te_d or {}).get("consumer_credit_thbmn"),
|
||
"household_debt_gdp": (te_d or {}).get("household_debt_gdp_pct"),
|
||
"consumer_confidence": (te_d or {}).get("consumer_confidence")},
|
||
"exploration": {"source": "TOP energy (refining)", "frequency": "quarterly",
|
||
"energy_proxy": surprises.get("refining_energy")},
|
||
}
|
||
for tid, read in proxy_reads.items():
|
||
themes.append(self._mk_theme(tid, surprises.get(tid), read))
|
||
|
||
# 4) combined board (60/40) via themes scoring
|
||
board = self._build_board(themes, macro_d)
|
||
|
||
# 5) source provenance table
|
||
sources = self._build_sources(auto_d, npl_d, en_d, macro_d, tourism, bnpl_d, trade_d, te_d, irpc_d)
|
||
|
||
return {
|
||
"themes": themes,
|
||
"macro": macro_d,
|
||
"board": board,
|
||
"sources": sources,
|
||
# unambiguous split so "7 vs 5" style confusion is impossible:
|
||
# distinct provider rows vs raw FACTORS-registry factor keys.
|
||
"source_summary": {
|
||
"rows": len(sources),
|
||
"factor_keys": _factor_key_count(),
|
||
},
|
||
"as_of": macro_d.get("periods", {}).get("headline_inflation_yoy", ""),
|
||
}
|
||
|
||
def _mk_theme(self, tid: str, surprise: Optional[float], read: dict) -> dict:
|
||
label = themes_mod.THEME_LABELS_TH.get(tid, tid)
|
||
f = themes_mod.THEME_FREQUENCY.get(tid, "monthly")
|
||
if isinstance(read, dict):
|
||
thesis = read.get("thesis", "")
|
||
read = {k: v for k, v in read.items() if k != "thesis"}
|
||
else:
|
||
thesis = ""
|
||
narrative = self._theme_narrative(tid, surprise, read)
|
||
return {
|
||
"id": tid, "label_th": label, "frequency": f,
|
||
"surprise": surprise, "read": read, "thesis": thesis, "narrative": narrative,
|
||
}
|
||
|
||
def _theme_narrative(self, tid: str, surprise: Optional[float], read: dict) -> str:
|
||
sign = "ดีขึ้น" if (surprise or 0) >= 0 else "แย่ลง"
|
||
s = f"{abs(surprise):.2f}σ" if surprise is not None else "—"
|
||
if tid == "auto_credit":
|
||
yoy = read.get("new_car_sales_yoy")
|
||
npl = read.get("auto_npl_pct")
|
||
yoy_txt = f"ยอดขายรถยนต์โต {yoy:+.1f}% เมื่อเทียบรายปี" if yoy is not None else "ยอดขายรถยนต์ไม่ชัดเจน"
|
||
npl_txt = f"สัดส่วนหนี้เสียรถยนต์ (NPL) อยู่ที่ {npl:.1f}%" if npl is not None else "ตัวเลขหนี้เสียรถยนต์ยังไม่ชัดเจน"
|
||
direction = ("ส่งผลบวกต่อกำลังซื้อรถยนต์และธุรกิจที่เกี่ยวข้อง" if (surprise or 0) >= 0
|
||
else "อาจกดดันกำไรของกลุ่มลิสซิ่ง/สินเชื่อรถ เพราะความสามารถชำระหนี้แย่ลง")
|
||
return (f"{yoy_txt} ขณะที่ {npl_txt}. ค่าความต่างรวม {s} บ่งชี้ทิศทาง{ ('ที่ดี' if (surprise or 0) >= 0 else 'ที่ต้องระวัง') } — "
|
||
f"{direction}. หุ้นที่พึ่งพารายได้จากรถยนต์/สินเชื่อรถ เช่น ลิสซิ่ง ธนาคารในกลุ่ม "
|
||
f"จะได้หรือเสียประโยชน์ตามทิศทางนี้.")
|
||
if tid == "refining_energy":
|
||
return (f"ค่าความต่าง {s} สำหรับธีมโรงกลั่น/พลังงาน "
|
||
f"{('สะท้อนกำไรขั้นต้นโรงกลั่นที่แข็งแรง' if (surprise or 0) >= 0 else 'สะท้อนแรงกดดันต่อกำไรโรงกลั่น')} "
|
||
f"จากข้อมูล TOP รายไตรมาส. กลุ่มพลังงาน (PTT, PTTGC, TOP, BCP, IRPC) จะได้รับผลตาม "
|
||
f"ทิศทางราคาพลังงานและค่าการกลั่น.")
|
||
if tid == "tourism":
|
||
return (f"ค่าความต่าง {s} สำหรับธีมการท่องเที่ยว "
|
||
f"{('บ่งชี้การท่องเที่ยวที่คึกคักกว่าปกติ' if (surprise or 0) >= 0 else 'บ่งชี้การท่องเที่ยวที่ซบเซากว่าปกติ')} "
|
||
f"จากข้อมูลการท่องเที่ยวประเทศ. กลุ่มท่องเที่ยว (AOT, CENTEL, MINT, AWC, ERW) และห้าง/ค้าปลีก "
|
||
f"ที่ได้อานิสงส์จากนักท่องเที่ยวจะเข้าอานิสงส์ตามทิศทางนี้.")
|
||
# --- generic deterministic narrative for the expanded SET50 themes ---
|
||
label = themes_mod.THEME_LABELS_TH.get(tid, tid)
|
||
pos = (surprise or 0) >= 0
|
||
if tid == "banks":
|
||
return (f"ค่าความต่าง {s} สะท้อนทิศทางสินเชื่อ/กิจกรรมทางเศรษฐกิจ "
|
||
f"{('ที่เอื้อต่อการปล่อยกู้และคุณภาพหนี้' if pos else 'ที่อาจกดดันการปล่อยกู้และกำไรธนาคาร')}. "
|
||
f"กลุ่มธนาคาร (BBL, KBANK, KTB, SCB, TTB) จะได้หรือเสียตามแรงส่งนี้.")
|
||
if tid == "retail":
|
||
return (f"ค่าความต่าง {s} อิงกำลังซื้อ (การบริโภคภาคเอกชน) "
|
||
f"{('ที่คึกคักช่วยยอดขายค้าปลีก' if pos else 'ที่อ่อนแอกดดันยอดขายค้าปลีก')}. "
|
||
f"กลุ่มค้าปลีก (CPALL, COM7, GLOBAL, HMPRO, OR, OSP) รับผลตามทิศทางนี้.")
|
||
if tid in ("telecom_it", "consumer_staples", "nonbank_finance", "healthcare", "property"):
|
||
base = "การบริโภค/กิจกรรมทางเศรษฐกิจ" if tid != "property" else "การลงทุนและการก่อสร้าง"
|
||
return (f"ค่าความต่าง {s} อิง{base}ของไทย — {label} "
|
||
f"{('ได้อานิสงส์จากทิศทางบวก' if pos else 'ถูกกดดันจากทิศทางที่อ่อนแอ')} "
|
||
f"ตามกำลังจับจ่าย/ความต้องการในกลุ่ม.")
|
||
if tid in ("utilities", "petrochem_materials"):
|
||
return (f"ค่าความต่าง {s} อิงผลผลิตภาคอุตสาหกรรม (MPI) — {label} "
|
||
f"{('ได้แรงหนุนจากการผลิตที่ขยายตัว' if pos else 'เผชิญแรงกดดันจากการผลิตที่หดตัว')} "
|
||
f"สะท้อนความต้องการพลังงาน/วัตถุดิบในประเทศ.")
|
||
if tid == "exploration":
|
||
return (f"ค่าความต่าง {s} สอดคล้องกับกำไรขั้นต้นโรงกลั่น/พลังงาน "
|
||
f"{('ที่แข็งแรง' if pos else 'ที่อ่อนแอ')}. กลุ่มสำรวจ-ผลิต (PTTEP, PTT) "
|
||
f"รับผลตามราคาพลังงานและค่าการกลั่น.")
|
||
return ""
|
||
|
||
def _tourism_surprise(self) -> Optional[float]:
|
||
"""Derive the tourism surprise from the bot-tourism observation set
|
||
(cross-sectional mean of signal scores), when available. Returns None
|
||
so `compute_theme_surprises` falls back to the registry factors."""
|
||
import statistics
|
||
ts = self.tourism_signals
|
||
if not ts:
|
||
return None
|
||
surprises = [x.get("score", 0) for x in ts if isinstance(x, dict)]
|
||
return round(statistics.mean(surprises), 3) if surprises else None
|
||
|
||
def _build_board(self, themes, macro_d) -> list:
|
||
# combine theme scores + siamchart for the per-symbol board.
|
||
from . import siamchart_factors
|
||
fv = siamchart_factors.build_factor_view() or {"factors": []}
|
||
momentum = themes_mod._load_momentum()
|
||
siamchart_score = themes_mod.build_siamchart_score(fv, momentum=momentum)
|
||
# per-theme symbol exposure: surprise × firm_quality (real selection).
|
||
# A strong name in a hot theme scores higher than a weak one.
|
||
theme_scores: dict[str, dict[str, float]] = {}
|
||
fv_all = fv
|
||
for t in themes:
|
||
tid = t["id"]
|
||
surprise = t.get("surprise")
|
||
if surprise is None:
|
||
theme_scores[tid] = {}
|
||
continue
|
||
symbols = themes_mod.THEME_SYMBOLS.get(tid, set())
|
||
q = {}
|
||
for s in symbols:
|
||
if s not in siamchart_score:
|
||
continue
|
||
quality = themes_mod.quality_within_theme(s, tid, fv_all)
|
||
q[s] = float(surprise) * quality
|
||
theme_scores[tid] = q
|
||
combined = themes_mod.combine_score(
|
||
list(theme_scores.values()), siamchart_score,
|
||
)
|
||
# factor metadata per symbol for the board columns
|
||
fmap = {f.get("symbol"): f for f in fv.get("factors", [])}
|
||
board = []
|
||
for sym, meta in combined.items():
|
||
f = fmap.get(sym, {})
|
||
# which themes this symbol belongs to (from THEME_SYMBOLS) — the
|
||
# frontend derives the theme column from this, never a local map.
|
||
sym_themes = [tid for tid, syms in themes_mod.THEME_SYMBOLS.items()
|
||
if sym in syms]
|
||
board.append({
|
||
"symbol": sym,
|
||
"combined": round(meta.get("combined", 0.0), 3),
|
||
"theme_score": round(meta.get("theme_score", 0.0), 3),
|
||
"siamchart_score": round(meta.get("siamchart_score", 0.0), 3),
|
||
"themes": sym_themes,
|
||
"dividend_yield": f.get("dividend_yield"),
|
||
"is_dividend": f.get("is_dividend"),
|
||
})
|
||
board.sort(key=lambda r: r["combined"], reverse=True)
|
||
return board
|
||
|
||
def _build_sources(self, auto_d, npl_d, en_d, macro_d, tourism, bnpl_d=None, trade_d=None, te_d=None, irpc_d=None) -> list:
|
||
"""Sources derived from the FACTORS registry — adding a factor to
|
||
factors.py auto-appends its source row here (no hardcoded list)."""
|
||
import datetime as _dt
|
||
from app import factors as factors_mod
|
||
now = _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="minutes")
|
||
|
||
fetched = {
|
||
"auto_credit": auto_d, "auto_npl": npl_d,
|
||
"energy_thai": en_d, "energy_irpc": irpc_d,
|
||
"macro_thai": macro_d, "bank_npl": bnpl_d,
|
||
"thai_trade": trade_d, "te_thailand": te_d,
|
||
}
|
||
# group FACTORS by fetch module -> one row per distinct source
|
||
source_by_module: dict = {}
|
||
for fkey, f in factors_mod.FACTORS.items():
|
||
mod = f.get("fetch")
|
||
if not mod:
|
||
continue
|
||
if mod not in source_by_module:
|
||
source_by_module[mod] = {
|
||
"mod": mod,
|
||
"source": f.get("source", mod),
|
||
"freq": f.get("frequency", "monthly"),
|
||
"factors": [],
|
||
}
|
||
source_by_module[mod]["factors"].append(fkey)
|
||
|
||
# next-update cadence per frequency (hours)
|
||
_cadence = {"daily": 24, "monthly": 24 * 30, "quarterly": 24 * 91,
|
||
"annual": 24 * 365, "weekly": 24 * 7}
|
||
|
||
rows = []
|
||
for mod, meta in source_by_module.items():
|
||
data = fetched.get(mod) or {}
|
||
_source_label = {
|
||
"auto_credit": "TradingEconomics",
|
||
"auto_npl": "BOT FI_NP_003_S2",
|
||
"bank_npl": "BOT FI_NP_003_S2",
|
||
"energy_thai": "Thai Oil investor",
|
||
"energy_irpc": "IRPC investor",
|
||
"macro_thai": "BOT Thai Economy",
|
||
"bot_tourism": "BOT Tourism",
|
||
"thai_trade": "TradingEconomics",
|
||
"te_thailand": "TradingEconomics",
|
||
}
|
||
freq = meta["freq"]
|
||
as_of = data.get("as_of") or data.get("period") or _period(data, meta["factors"])
|
||
next_in_h = _cadence.get(freq, 24 * 30)
|
||
next_at = (now_plus(now, next_in_h))
|
||
rows.append({
|
||
"แหล่ง": _source_label.get(mod, meta["source"]),
|
||
"ขอบเขต": " | ".join(factors_mod.FACTORS[k]["name_th"] for k in meta["factors"]),
|
||
"ข้อมูล": as_of or "",
|
||
"ความถี่": freq_th(freq),
|
||
"อัปเดตครั้งต่อไป": next_at,
|
||
"dึงมาเมื่อ": now,
|
||
})
|
||
# tourism (fetched separately)
|
||
if tourism and isinstance(tourism, dict):
|
||
rows.insert(0, {
|
||
"แหล่ง": "BOT Tourism",
|
||
"ขอบเขต": "นักท่องเที่ยว",
|
||
"ข้อมูล": tourism.get("period", "") or "",
|
||
"ความถี่": "รายเดือน",
|
||
"อัปเดตครั้งต่อไป": now_plus(now, 24 * 30),
|
||
"dึงมาเมื่อ": now,
|
||
})
|
||
return rows
|
||
|
||
|
||
def now_plus(iso_now: str, hours: float) -> str:
|
||
import datetime as _dt
|
||
base = _dt.datetime.fromisoformat(iso_now)
|
||
return (base + _dt.timedelta(hours=hours)).isoformat(timespec="minutes")
|
||
|
||
|
||
def freq_th(freq: str) -> str:
|
||
return {"daily": "รายวัน", "weekly": "รายสัปดาห์", "monthly": "รายเดือน",
|
||
"quarterly": "รายไตรมาส", "annual": "รายปี"}.get(freq, freq)
|
||
|
||
|
||
def _period(data: dict, factor_keys: list) -> str:
|
||
from app import factors as factors_mod
|
||
for k in factor_keys:
|
||
f = factors_mod.FACTORS.get(k, {})
|
||
vk = f.get("value_key")
|
||
if vk and data.get(vk) is not None:
|
||
return f.get("name_th", k)
|
||
return "--"
|
||
|
||
|
||
def _factor_key_count() -> int:
|
||
"""Number of FACTORS-registry entries (each a distinct factor key)."""
|
||
from app import factors as factors_mod
|
||
return len(factors_mod.FACTORS)
|
||
|
||
|
||
def default_scores(syms: Optional[list] = None) -> dict:
|
||
"""Per-symbol {combined, is_dividend, dividend_yield} from the live board.
|
||
|
||
Used as the default baseline for the backtest & simulation engines (honest:
|
||
current combined scores; a PIT score_fn can be supplied to avoid lookahead).
|
||
`syms=None` returns every symbol on the board.
|
||
"""
|
||
from app import daily_cache
|
||
from app import siamchart_factors
|
||
fv = siamchart_factors.build_factor_view()
|
||
cache = daily_cache.DailyCache()
|
||
dash = RealDashboard([], cache, factor_view=fv).build()
|
||
out = {}
|
||
for row in dash.get("board", []):
|
||
out[row["symbol"]] = {
|
||
"combined": row.get("combined", 0.0),
|
||
"is_dividend": row.get("is_dividend", False),
|
||
"dividend_yield": row.get("dividend_yield") or 0.0,
|
||
}
|
||
if syms:
|
||
out = {s: out.get(s, {}) for s in syms if s in out}
|
||
return out
|