129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
"""录入与导入：人工 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()),
|
||
}
|