"""录入与导入:人工 CSV 导入 + 自动化接入接口(预留)。 - parse_events_csv / parse_etf_csv:解析导入模板(见 ddl/*_template.csv) - AutoEventSource / AutoEtfSource:自动化抓取接口(未来挂新闻/公告/资金流), 现在留空注册表;实现后在 _AUTO_* 注册即可被定时任务调用。 """ from __future__ import annotations import csv import io import logging from datetime import date, datetime from typing import Dict, List, Protocol log = logging.getLogger("as-event.importers") _VALID_DIRECTION = {"bullish", "bearish", "neutral"} _VALID_CYCLE = {"top", "bottom", "none"} def _parse_date(s: str) -> date: s = (s or "").strip() for fmt in ("%Y%m%d", "%Y-%m-%d", "%Y/%m/%d"): try: return datetime.strptime(s, fmt).date() except ValueError: continue raise ValueError(f"日期格式错误: {s!r}") def _clean(s) -> str: return (s or "").strip() def parse_events_csv(content: str) -> List[dict]: """列:event_date,title,category,impact_direction,severity,related_indices,cycle_tag,description,source_url""" reader = csv.DictReader(io.StringIO(content)) out: List[dict] = [] for i, row in enumerate(reader, start=2): # 含表头,数据从第2行 title = _clean(row.get("title")) if not title: continue try: ev = dict( event_date=_parse_date(row.get("event_date", "")), title=title, category=_clean(row.get("category")) or "其他", impact_direction=(_clean(row.get("impact_direction")).lower() or "neutral"), severity=int(_clean(row.get("severity")) or 3), related_indices=_clean(row.get("related_indices")) or None, cycle_tag=(_clean(row.get("cycle_tag")).lower() or "none"), description=_clean(row.get("description")) or None, source_url=_clean(row.get("source_url")) or None, source="import", ) except ValueError as e: raise ValueError(f"第 {i} 行解析失败: {e}") from e if ev["impact_direction"] not in _VALID_DIRECTION: ev["impact_direction"] = "neutral" if ev["cycle_tag"] not in _VALID_CYCLE: ev["cycle_tag"] = "none" ev["severity"] = min(5, max(1, ev["severity"])) out.append(ev) return out def parse_etf_csv(content: str) -> List[dict]: """列:trade_date,etf_code,etf_name,category,related_index,net_inflow,shares_change,note""" reader = csv.DictReader(io.StringIO(content)) out: List[dict] = [] for i, row in enumerate(reader, start=2): d = _clean(row.get("trade_date")) if not d: continue try: rec = dict( trade_date=_parse_date(d), etf_code=_clean(row.get("etf_code")), etf_name=_clean(row.get("etf_name")) or None, category=_clean(row.get("category")) or None, related_index=_clean(row.get("related_index")) or None, net_inflow=float(_clean(row.get("net_inflow")) or 0) if _clean(row.get("net_inflow")) else None, shares_change=float(_clean(row.get("shares_change"))) if _clean(row.get("shares_change")) else None, note=_clean(row.get("note")) or None, source="import", ) except ValueError as e: raise ValueError(f"第 {i} 行解析失败: {e}") from e out.append(rec) return out # ============================ 自动化接入接口(预留)============================ class AutoEventSource(Protocol): """未来自动抓取事件源的接口。实现后注册到 _AUTO_EVENT_SOURCES。""" name: str def pull(self, start: date, end: date) -> List[dict]: # 返回同 parse_events_csv 的 dict 列表 ... class AutoEtfSource(Protocol): name: str def pull(self, start: date, end: date) -> List[dict]: ... _AUTO_EVENT_SOURCES: Dict[str, AutoEventSource] = {} _AUTO_ETF_SOURCES: Dict[str, AutoEtfSource] = {} def register_auto_event_source(src: AutoEventSource) -> None: _AUTO_EVENT_SOURCES[src.name] = src log.info("注册自动事件源: %s", src.name) def register_auto_etf_source(src: AutoEtfSource) -> None: _AUTO_ETF_SOURCES[src.name] = src log.info("注册自动ETF源: %s", src.name) def list_auto_sources() -> Dict[str, List[str]]: return { "event": list(_AUTO_EVENT_SOURCES.keys()), "etf": list(_AUTO_ETF_SOURCES.keys()), }