commit 2c72de1dba1ae6052d97710fcacbe5a71b354ac4 Author: lookt Date: Tue Sep 15 20:04:13 2026 +0800 windows-desktop: C端基线（Windows 本地实时分析器 + 采集器 bug 修复） diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..977a7d0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +data/ +notebooks/.ipynb_checkpoints/ +*.log +.a_stock_timeline/ diff --git a/docs/DATA_DICTIONARY.md b/docs/DATA_DICTIONARY.md new file mode 100644 index 0000000..572e2bd --- /dev/null +++ b/docs/DATA_DICTIONARY.md @@ -0,0 +1,358 @@ +# A 股事件-市场时间线分析工具链：数据字典 + +> 项目路径：`C:\Users\lookt\a_stock_timeline` +> 数据库：`data/a_stock.db`（SQLite） +> 数据导出：`data/a_stock_all.xlsx`（8 个 Sheet） + +--- + +## 一、整体架构 + +``` +数据采集层（src/fetcher/） + │ + ▼ +数据持久层（data/a_stock.db — 8 张表） + │ + ├──► CSV 备份（data/raw/） + │ + ▼ +时间线对齐层（src/align/） + │ + ▼ +可视化层（src/viz/） +``` + +--- + +## 二、各表详细说明 + +--- + +### 2.1 `trade_calendar` —— 交易日历 + +| 属性 | 说明 | +|---|---| +| 行数 | 8,797 | +| 时间范围 | 1990-12-19 至今 | +| 数据源 | AkShare `tool_trade_date_hist_sina` | +| 用途 | 判断某日是否为交易日；时间线对齐的锚点 | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `trade_date` | TEXT（YYYY-MM-DD） | 日期 | +| `is_trade_day` | INTEGER（0/1） | 1=交易日，0=非交易日（周末/节假日） | + +**使用示例：** +- `WHERE is_trade_day = 1` 筛选实际交易日 +- 与 `stock_daily` 做 JOIN，确认某日是否有指数数据 +- 用于快讯/新闻时间戳补全（若时间戳缺失日期，默认填最近交易日） + +--- + +### 2.2 `stock_daily` —— A 股主要指数日线行情 + +| 属性 | 说明 | +|---|---| +| 行数 | 315 | +| 指数数量 | 5 只（上证、深证、沪深300、创业板、科创50） | +| 时间范围 | 近 90 个交易日 | +| 数据源 | AkShare `index_zh_a_hist` | +| 用途 | 反映大盘整体走势，是时间线的"市场背景层" | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `ts_code` | TEXT | 指数代码（sh000001=上证综指，sz399001=深证成指，sh000300=沪深300，sz399006=创业板指，sh000688=科创50） | +| `trade_date` | TEXT（YYYY-MM-DD） | 交易日期 | +| `open` | REAL | 开盘点位 | +| `high` | REAL | 最高点位 | +| `low` | REAL | 最低点位 | +| `close` | REAL | 收盘点位 | +| `volume` | REAL | 成交量（股） | +| `amount` | REAL（可能为 NULL） | 成交额（元），部分指数无成交额数据 | + +**指数代码速查：** + +| ts_code | 名称 | +|---|---| +| sh000001 | 上证综指 | +| sz399001 | 深证成指 | +| sh000300 | 沪深300 | +| sz399006 | 创业板指 | +| sh000688 | 科创50 | + +**分析提示：** +- 配合 `pct_change = (close - prev_close) / prev_close * 100` 计算日收益率 +- 与宏观事件 JOIN，观察重大政策/数据发布前后指数的短期反应 +- 与龙虎榜 JOIN，识别"指数下跌 + 游资活跃"的异常交易日 + +--- + +### 2.3 `lhb_daily` —— 龙虎榜明细 + +| 属性 | 说明 | +|---|---| +| 行数 | 409 | +| 时间范围 | 近 90 个交易日 | +| 数据源 | AkShare `stock_lhb_detail_daily_sina` | +| 用途 | 追踪活跃游资席位的操作标的，识别短线热点 | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `trade_date` | TEXT（YYYY-MM-DD） | 上榜日期 | +| `ts_code` | TEXT | 股票代码 | +| `name` | TEXT | 股票名称（可能有编码问题，实际显示正常） | +| `close` | REAL | 上榜日收盘价 | +| `pct_change` | REAL | 当日涨跌幅（%） | +| `amount` | REAL | 龙虎榜成交金额（万元） | +| `reason` | TEXT | 上榜原因（如：涨跌幅偏离值达7%、日换手率达20%等） | + +**常见上榜原因含义：** + +| reason 关键词 | 含义 | +|---|---| +| 涨跌幅偏离值达7% | 单日涨幅或跌幅过大，被交易所监控 | +| 日换手率达20% | 筹码换手频繁，可能是游资接力 | +| 连续3个交易日内涨跌幅偏离值累计达20% | 短线连板股 | +| 日振幅达15% | 日内波动剧烈，多空分歧大 | +| 异常波动 | 无明显原因的异动 | + +**分析提示：** +- 按 `reason` 分组，统计近期哪类上榜原因最频繁 +- 结合指数涨跌（`stock_daily`）判断：大盘跌时仍有大量涨停，可能是个股独立行情 +- 高频上榜股（同一股票多日出现）往往是市场主线龙头 + +--- + +### 2.4 `news_flash` —— 财经快讯流 + +| 属性 | 说明 | +|---|---| +| 行数 | 10（演示数据，批量采集后可增至数千条/日） | +| 时间精度 | 分钟级 | +| 数据源 | EastMoney `np-anotice-stock`（主），AkShare `stock_news_em`（降级） | +| 用途 | 实时捕获影响市场的突发消息，作为时间线的"事件触发层" | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `content` | TEXT | 快讯正文内容（可能有编码问题） | +| `event_time` | TEXT（YYYY-MM-DD HH:MM:SS） | 事件发生时间 | +| `source` | TEXT | 来源（如：东方财富Choice数据、证券时报） | +| `inserted_at` | TEXT（YYYY-MM-DD HH:MM:SS） | 入库时间 | + +**分析提示：** +- 快讯内容可能包含：政策发布、经济数据、监管动态、个股公告 +- 可按关键词（"降息"、"关税"、"制裁"、"收购"）做规则过滤 +- 高频来源（东方财富、同花顺）可作为可信度参考 + +--- + +### 2.5 `macro_event` —— 宏观经济事件日历 + +| 属性 | 说明 | +|---|---| +| 行数 | 99 | +| 数据源 | AkShare `news_economic_baidu`（财经日历） | +| 用途 | 追踪全球主要经济数据发布时间（如非农、CPI、利率决议） | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `event_date` | TEXT（YYYY-MM-DD） | 事件公布日期 | +| `title` | TEXT | 事件标题（当前入库存在字段映射问题，详见下方警告） | +| `event_type` | INTEGER | 重要性等级（1=高，2=中，3=低） | +| `source` | TEXT | 数据来源 | +| `inserted_at` | TEXT（YYYY-MM-DD HH:MM:SS） | 入库时间 | + +> **警告（已知问题）：** 当前入库的 `title` 字段实际包含了原始数据的 `event_date`，而真正的标题内容写入了 `event_date` 字段。这是 AkShare 返回字段名与实际位置不一致导致的映射错误。修复方案：调整 `macro_event_fetcher.py` 中的字段映射逻辑，按返回数据的实际列位置重新对应。 + +**常见宏观事件类型：** + +| 事件 | 典型影响 | +|---|---| +| 美国非农就业数据 | 超预期=美元走强，A股承压 | +| CPI/PPI 数据 | 超预期=通胀担忧，货币政策预期收紧 | +| 美联储利率决议 | 降息=全球流动性宽松，A股受益 | +| 中国官方PMI | 超预期=经济复苏预期增强 | +| 进出口数据 | 超预期=外需强劲 | + +--- + +### 2.6 `news_cn` —— A 股市场新闻 + +| 属性 | 说明 | +|---|---| +| 行数 | 10（演示数据） | +| 数据源 | AkShare `stock_news_em` | +| 用途 | 覆盖个股/行业/宏观层面的新闻报道，构建事件叙事 | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `id` | INTEGER | 自增主键 | +| `title` | TEXT | 新闻标题 | +| `content` | TEXT | 新闻正文摘要 | +| `pub_date` | TEXT（YYYY-MM-DD HH:MM:SS） | 发布时间 | +| `source` | TEXT | 来源媒体 | +| `url` | TEXT | 原文链接 | +| `keywords` | TEXT | 关键词标签（逗号分隔） | +| `inserted_at` | TEXT（YYYY-MM-DD HH:MM:SS） | 入库时间 | + +--- + +### 2.7 `money_flow` —— 个股资金流向（主力净流入） + +| 属性 | 说明 | +|---|---| +| 行数 | 2,085 | +| 时间范围 | 最近 1 个交易日（全市场个股） | +| 数据源 | AkShare `stock_fund_flow_individual` | +| 用途 | 追踪主力资金（超大单+大单）的每日净流入/出，识别资金轮动方向 | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `ts_code` | TEXT | 股票代码 | +| `trade_date` | TEXT（YYYY-MM-DD） | 交易日期 | +| `main_net_in` | REAL（当前全为 NULL，待修复） | 主力净流入金额（元） | + +> **已知问题：** `main_net_in` 字段当前全部为 NULL。原因是 AkShare `stock_fund_flow_individual` 接口返回的数值包含"万"、"亿"等中文单位后缀，字符串转数值时解析失败。修复方案：在 `fund_flow_fetcher.py` 中先对单位字符串进行清洗（去掉"万"/"亿"，按需转换精度），再转为浮点数。 + +**资金流向分析逻辑：** + +| 主力净流入方向 | 市场含义 | +|---|---| +| 连续多日净流入某板块 | 机构建仓，可能是主线行情启动 | +| 大盘下跌但某股净流入 | 护盘资金或逆势买入 | +| 大盘上涨但某股净流出 | 主力借机减仓 | + +--- + +### 2.8 `global_index` —— 外盘重要指数 + +| 属性 | 说明 | +|---|---| +| 行数 | 14,583 | +| 时间范围 | 道琼斯（2004-至今）、纳斯达克100（2004-至今）、标普500（2004-至今） | +| 数据源 | AkShare `index_us_stock_sina` | +| 用途 | 反映美股走势，作为 A 股的"外围环境层"（隔夜美股涨跌影响次日 A 股情绪） | + +**字段：** + +| 字段名 | 类型 | 说明 | +|---|---|---| +| `index_code` | TEXT | 指数代码（US.DJI=道琼斯，US.NDX=纳斯达克100，US.INX=标普500） | +| `trade_date` | TEXT（YYYY-MM-DD） | 交易日期 | +| `close` | REAL | 收盘点位 | +| `pct_change` | REAL | 当日涨跌幅（%），由收盘价自计算 | + +**外盘与 A 股的联动规律：** + +| 场景 | A 股次日预期 | +|---|---| +| 美股大涨（纳指+2%以上） | 高开概率大 | +| 美股大跌（道指-2%以上） | A 股低开，但可能低开高走 | +| 美股窄幅震荡 | A 股按自身逻辑运行 | +| 美股期货夜盘异动 | 可作为盘前情绪参考 | + +--- + +## 三、数据关联关系 + +``` +trade_calendar（锚点） + │ + ├──► stock_daily ON trade_date（大盘指数基准） + ├──► lhb_daily ON trade_date（游资活跃度） + ├──► money_flow ON trade_date（资金流向） + │ +news_flash ON event_time（分钟级事件） + │ +news_cn ON pub_date（新闻发布日） + │ +macro_event ON event_date（宏观事件日） + │ +global_index ON trade_date-1（美股影响传导，差一天） +``` + +**时间线对齐策略：** +1. 以 `trade_date`（交易日）为基准轴 +2. 快讯/新闻精确到分钟，非交易日数据合并到最近交易日 +3. 宏观事件按公布日期对齐 +4. 外盘指数按"美股T日 → A股T+1日"的逻辑传导对齐 + +--- + +## 四、数据质量问题汇总 + +| 问题 | 涉及表 | 严重程度 | 状态 | +|---|---|---|---| +| `main_net_in` 全为 NULL | `money_flow` | 高 | 待修复 | +| 字段映射错位 | `macro_event` | 高 | 待修复并重新入库 | +| 中文内容乱码（GBK/UTF-8） | `lhb_daily`, `news_flash`, `news_cn` | 低（数据本身正确，仅显示问题） | 排查终端编码 | +| 采集时间范围受限 | `stock_daily` | 中 | 可扩展至更长历史 | +| `macro_event` 为历史数据非未来事件 | `macro_event` | 中 | 需配置 AkShare 实时财经日历参数 | + +--- + +## 五、快速查询示例 + +```sql +-- 查最近 5 个交易日的大盘涨跌 +SELECT trade_date, close, + ROUND((close - LAG(close) OVER(ORDER BY trade_date)) / LAG(close) OVER(ORDER BY trade_date) * 100, 2) AS pct_change +FROM stock_daily +WHERE ts_code = 'sh000001' +ORDER BY trade_date DESC LIMIT 5; + +-- 查近 30 天龙虎榜热门上榜原因 +SELECT reason, COUNT(*) AS cnt +FROM lhb_daily +WHERE trade_date >= date('now', '-30 days') +GROUP BY reason +ORDER BY cnt DESC LIMIT 10; + +-- 查某日主力净流入TOP10个股 +SELECT ts_code, main_net_in +FROM money_flow +WHERE trade_date = '2026-09-15' +ORDER BY main_net_in DESC NULLS LAST LIMIT 10; + +-- 查近 7 天重大宏观事件 +SELECT event_date, title +FROM macro_event +WHERE event_date >= date('now', '-7 days') +ORDER BY event_date; +``` + +--- + +## 六、后续数据扩展方向 + +| 优先级 | 数据类型 | 数据源 | 说明 | +|---|---|---|---| +| P0 | 资金流向（修复） | 修复 `fund_flow_fetcher.py` 逻辑 | 解决 NULL 问题 | +| P0 | 宏观事件（修复） | 修复 `macro_event_fetcher.py` 映射 | 解决字段错位 | +| P1 | 港股行情 | AkShare `index_hk_hist` | 补充港股恒生指数 | +| P1 | 期货行情 | AkShare `futures_zh_daily_sina` | 商品期货（黄金/原油） | +| P1 | 北向资金 | AkShare `stock_hsgt_north_net_flow_in` | 外资流向 | +| P2 | 行业板块涨跌 | AkShare `stock_board_industry_name_em` | 板块轮动 | +| P2 | 融资融券 | AkShare `stock_margin_detail_sz` | 杠杆资金动向 | +| P2 | 恐慌指数（VIX） | AkShare | 市场情绪反向指标 | + +--- + +*文档版本：v1.0* +*最后更新：2026-09-15* +*对应数据库快照：`data/a_stock.db`（8 表）* diff --git a/export_all.py b/export_all.py new file mode 100644 index 0000000..03968b9 --- /dev/null +++ b/export_all.py @@ -0,0 +1,30 @@ +"""全库导出为 xlsx（每个表一个 sheet）""" +import sqlite3 +import pandas as pd +from pathlib import Path +from datetime import datetime + +DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db' +OUT = r'C:\Users\lookt\a_stock_timeline\data\a_stock_all.xlsx' + +TABLES = [ + 'trade_calendar', + 'stock_daily', + 'lhb_daily', + 'news_flash', + 'macro_event', + 'news_cn', + 'money_flow', + 'global_index', +] + +conn = sqlite3.connect(DB) + +with pd.ExcelWriter(OUT, engine='openpyxl', datetime_format='YYYY-MM-DD') as writer: + for tbl in TABLES: + df = pd.read_sql(f'SELECT * FROM {tbl}', conn) + print(f'{tbl}: {len(df)} 行 -> sheet') + df.to_excel(writer, sheet_name=tbl, index=False) + +conn.close() +print(f'\n[OK] 导出完成: {OUT}') diff --git a/fix_fetchers.py b/fix_fetchers.py new file mode 100644 index 0000000..a4189d8 --- /dev/null +++ b/fix_fetchers.py @@ -0,0 +1,151 @@ +# -*- coding: utf-8 -*- +"""修复 fetcher：fund_flow 单位乘数+保历史；macro 按列名映射+保历史；stock_api 全量历史+保历史""" +import io + +ROOT = r'C:\Users\lookt\a_stock_timeline' + +def read(p): + return io.open(p, encoding='utf-8').read() + +def write(p, s): + io.open(p, 'w', encoding='utf-8').write(s) + print('patched:', p.split('a_stock_timeline')[-1]) + +NL = chr(10) +UP = 'from src.storage.db import upsert_df, upsert_rows, upsert_log, get_last_fetch' +UP_OLD = 'from src.storage.db import upsert_df, upsert_log, get_last_fetch' + +# ── 2) fund_flow_fetcher ── +p = ROOT + r'\src\fetcher\fund_flow_fetcher.py' +s = read(p) +if 'def parse_cn_amount' not in s: + s = s.replace(UP_OLD, UP) + helper = NL.join([ + "DB = r'C:" + chr(92) + "Users" + chr(92) + "lookt" + chr(92) + "a_stock_timeline" + + chr(92) + "data" + chr(92) + "a_stock.db'", + "", + "", + "def parse_cn_amount(v):", + " '''1.5亿 -> 1.5e8, -300万 -> -3e6，逗号/空白容错；无效返回 None'''", + " if v is None or (isinstance(v, float) and v != v):", + " return None", + " s = str(v).replace(',', '').replace(' ', '').strip()", + " if not s or s in ('--', '-', 'nan'):", + " return None", + " mult = 1.0", + " if s.endswith('亿'):", + " mult, s = 1e8, s[:-1]", + " elif s.endswith('万'):", + " mult, s = 1e4, s[:-1]", + " try:", + " return float(s) * mult", + " except ValueError:", + " return None", + ""]) + marker = "DB = r'C:" + chr(92) + "Users" + chr(92) + "lookt" + chr(92) + "a_stock_timeline" + chr(92) + "data" + chr(92) + "a_stock.db'" + NL + s = s.replace(marker, helper, 1) + old = NL.join([ + " # 主力净流入转数值（去掉\"万\"/\"亿\"）", + " df_ind['main_net_in'] = (df_ind['main_net_in'].astype(str)", + " .str.replace(r'[\\u4e00-\\u9fa5万/亿]', '', regex=True))", + " df_ind['main_net_in'] = pd.to_numeric(df_ind['main_net_in'], errors='coerce')"]) + new = NL.join([ + " # 主力净流入转数值：亿=1e8 万=1e4（带乘数换算）", + " df_ind['main_net_in'] = df_ind['main_net_in'].map(parse_cn_amount)", + " df_ind = df_ind[df_ind['main_net_in'].notna()]"]) + if old in s: + s = s.replace(old, new) + old2 = NL.join([ + " df = pd.concat(rows, ignore_index=True)", + " conn = sqlite3.connect(DB)", + " df.to_sql('money_flow', conn, if_exists='replace', index=False)", + " conn.close()", + " log.info(f'写入 money_flow: {len(df)} 行')", + " return df"]) + new2 = NL.join([ + " df = pd.concat(rows, ignore_index=True)", + " need = ['ts_code', 'trade_date', 'main_net_in',", + " 'large_net_in', 'medium_net_in', 'small_net_in']", + " for c in need:", + " if c not in df.columns:", + " df[c] = None", + " n = upsert_rows(df[need], 'money_flow', conflict_cols=['ts_code', 'trade_date'])", + " log.info(f'写入 money_flow: {n} 行（历史保留）')", + " return df"]) + if old2 in s: + s = s.replace(old2, new2) + write(p, s) + +# ── 3) macro_event_fetcher ── +p = ROOT + r'\src\fetcher\macro_event_fetcher.py' +s = read(p) +s = s.replace(UP_OLD, UP) +old = " df.columns = [c.lower() for c in df.columns]" +new = NL.join([ + " # 实测列（akshare 1.18.92）: 日期, 时间, 地区, 事件, 公布, 预期, 前值, 重要性", + " # 按列【名称】映射，避免位置错位", + " col_map = {'日期': 'event_date', '时间': 'event_time', '地区': 'source',", + " '事件': 'title', '公布': 'actual', '预期': 'forecast',", + " '前值': 'prev', '重要性': 'event_type'}", + " df = df.rename(columns=col_map)"]) +if old in s: + s = s.replace(old, new, 1) +old2 = NL.join([ + ' df["event_type"] = df.get("importance", pd.Series([1] * len(df))).astype(int)', + ' df["source"] = df.get("country", pd.Series([""] * len(df))).astype(str)']) +new2 = NL.join([ + ' df["event_type"] = pd.to_numeric(df["event_type"], errors="coerce").fillna(0).astype(int)', + ' df["source"] = df["source"].fillna("").astype(str)']) +if old2 in s: + s = s.replace(old2, new2) +old3 = NL.join([ + 'def incremental_events(source: str = "macro_event") -> pd.DataFrame:', + ' """增量采集宏观事件"""', + ' # 优先 AkShare 财经日历（已知可用）', + ' df = fetch_macro_calendar()', + ' if df is not None and not df.empty:', + ' upsert_df(df, source)', + ' upsert_log(source, "", last_date=datetime.today().strftime("%Y-%m-%d"))', + ' return df']) +new3 = NL.join([ + 'def incremental_events(source: str = "macro_event") -> pd.DataFrame:', + ' """增量采集宏观事件（按 event_date+event_time+title 去重，历史保留）"""', + ' df = fetch_macro_calendar()', + ' if df is not None and not df.empty:', + " need = [c for c in ['event_date', 'event_time', 'title', 'event_type', 'source',", + " 'actual', 'forecast', 'prev', 'inserted_at'] if c in df.columns]", + " upsert_rows(df[need], source,", + " conflict_cols=['event_date', 'event_time', 'title'])", + ' upsert_log(source, "", last_date=datetime.today().strftime("%Y-%m-%d"))', + ' return df']) +if old3 in s: + s = s.replace(old3, new3) +write(p, s) + +# ── 4) stock_api ── +p = ROOT + r'\src\fetcher\stock_api.py' +s = read(p) +s = s.replace(UP_OLD, UP) +old = NL.join([ + " df = batch_fetch_indices(days=days)", + " if df is not None and not df.empty:", + ' upsert_df(df, "stock_daily")']) +new = NL.join([ + " df = batch_fetch_indices(days=days)", + " if df is not None and not df.empty:", + ' upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"])']) +if old in s: + s = s.replace(old, new) +old_b = NL.join([ + " df = fetch_daily(code, start, end)", + " if df is not None and not df.empty:", + ' upsert_df(df, "stock_daily")']) +new_b = NL.join([ + " df = fetch_daily(code, start, end)", + " if df is not None and not df.empty:", + ' upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"])']) +if old_b in s: + s = s.replace(old_b, new_b) +write(p, s) + +print('ALL FETCHER PATCHES DONE') diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..f44d565 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +# a_stock_timeline diff --git a/src/analysis/mine_patterns.py b/src/analysis/mine_patterns.py new file mode 100644 index 0000000..5455d0d --- /dev/null +++ b/src/analysis/mine_patterns.py @@ -0,0 +1,211 @@ +# -*- coding: utf-8 -*- +""" +A股时间线实证模式挖掘：从 a_stock.db 挖掘隔夜传导/跳空回补/周内效应/ +波动率机制/龙虎榜统计，输出 markdown 到 E:/Data/skills/a-stock-timeline-patterns/references/ +口径：上证/沪深300 主样本；隔夜传导以纳斯达克为主参照。 +""" +import sqlite3 +import warnings + +import numpy as np +import pandas as pd + +warnings.filterwarnings('ignore') + +DB = r'C:' + chr(92) + 'Users' + chr(92) + 'lookt' + chr(92) + 'a_stock_timeline' + chr(92) + 'data' + chr(92) + 'a_stock.db' +OUT = r'E:' + chr(92) + 'Data' + chr(92) + 'skills' + chr(92) + 'a-stock-timeline-patterns' + chr(92) + 'references' + +IDX_NAMES = {'sh000001': '上证指数', 'sz399001': '深证成指', 'sh000300': '沪深300', + 'sz399006': '创业板指', 'sh000688': '科创50'} + + +def load(): + conn = sqlite3.connect(DB) + daily = pd.read_sql("SELECT ts_code, trade_date, open, high, low, close, volume " + "FROM stock_daily ORDER BY ts_code, trade_date", conn) + daily['trade_date'] = pd.to_datetime(daily['trade_date']) + for c in ('open', 'high', 'low', 'close'): + daily[c] = pd.to_numeric(daily[c], errors='coerce') + daily = daily.dropna(subset=['open', 'close']) + daily = daily[daily['close'] > 0] + g = pd.read_sql("SELECT index_code, trade_date, close, pct_change FROM global_index", conn) + g['trade_date'] = pd.to_datetime(g['trade_date']) + lhb = pd.read_sql("SELECT * FROM lhb_daily", conn) + conn.close() + return daily, g, lhb + + +def enrich(daily): + """按指数补充：日收益/隔夜跳空/日内波动/已实现波动率""" + out = [] + for code, df in daily.groupby('ts_code'): + df = df.sort_values('trade_date').copy() + df['prev_close'] = df['close'].shift(1) + df['ret'] = df['close'] / df['prev_close'] - 1 + df['gap'] = df['open'] / df['prev_close'] - 1 + df['intra'] = df['close'] / df['open'] - 1 + df['ret20'] = df['ret'].rolling(20).std() + df['ret20_prev'] = df['ret20'].shift(20) + df['weekday'] = df['trade_date'].dt.dayofweek + out.append(df) + return pd.concat(out, ignore_index=True) + + +def fmt_pct(v, digits=2): + return "{:.{}f}%".format(v * 100, digits) + + +def stat_line(sub, col=None): + s = sub.dropna() if hasattr(sub, 'dropna') else sub + if len(s) == 0: + return "样本不足" + up = len(s[s > 0]) / len(s) * 100 + return "N={}, 均值 {}, 中位数 {}, 胜率 {:.1f}%".format( + len(s), fmt_pct(s.mean()), fmt_pct(s.median()), up) + + +def overnight_matrix(daily, g): + lines = ["## 隔夜传导矩阵（美股 -> A股）", "", + "> 口径：美股T日涨跌幅分箱 -> A股T+1交易日（跳空=开盘/前收-1；日内=收盘/开盘-1）。" + "样本 2005-2026（受 A 股指数历史与纳斯达克 2014 起数据限制）。", ""] + for us_code, us_name in [('US.NDX', '纳斯达克100'), ('US.SPX', '标普500'), ('US.DJI', '道琼斯')]: + gu = g[g['index_code'] == us_code][['trade_date', 'pct_change']].rename( + columns={'pct_change': 'us_ret'}) + if gu.empty: + continue + gu['us_ret'] = pd.to_numeric(gu['us_ret'], errors='coerce') / 100.0 + bins = [-np.inf, -0.015, -0.005, 0.005, 0.015, np.inf] + labels = ['跌>1.5%', '跌0.5~1.5%', '正负0.5%内', '涨0.5~1.5%', '涨>1.5%'] + gu['us_bin'] = pd.cut(gu['us_ret'], bins=bins, labels=labels) + for idx_code in ['sh000001', 'sh000300']: + a = daily[daily['ts_code'] == idx_code][['trade_date', 'gap', 'intra', 'ret']].copy() + a = a.sort_values('trade_date') + merged = pd.merge_asof(a, gu.sort_values('trade_date'), + left_on='trade_date', right_on='trade_date', + direction='backward', allow_exact_matches=False) + merged = merged.dropna(subset=['us_bin', 'gap']) + name = IDX_NAMES.get(idx_code, idx_code) + lines.append("### {}T日 -> {}T+1日".format(us_name, name)) + lines.append("") + for lab in labels: + sub = merged[merged['us_bin'] == lab] + if len(sub) < 30: + lines.append("- 美股{}: 样本不足（{}）".format(lab, len(sub))) + continue + lines.append("- 美股{}（{}天）: A股跳空 {}; 日内 {}; 全天 {}".format( + lab, len(sub), stat_line(sub['gap']), stat_line(sub['intra']), + stat_line(sub['ret']))) + lines.append("") + return chr(10).join(lines) + + +def gap_intraday(daily): + lines = ["## A股指数跳空回补统计（日外->日内的接力规律）", "", + "> 口径：跳空=开盘/前收-1；日内=收盘/开盘-1。日内为负即\"高开低走\"（回落/回补）。", ""] + bins = [(-np.inf, -0.01, '大幅低开<-1%'), (-0.01, -0.002, '低开0.2~1%'), + (-0.002, 0.002, '平开正负0.2%'), (0.002, 0.01, '高开0.2~1%'), + (0.01, np.inf, '大幅高开>1%')] + for idx_code in ['sh000001', 'sh000300']: + a = daily[daily['ts_code'] == idx_code].dropna(subset=['gap', 'intra']) + name = IDX_NAMES.get(idx_code, idx_code) + years = "{}-{}".format(a['trade_date'].dt.year.min(), a['trade_date'].dt.year.max()) + lines.append("### {}（{}，{}天）".format(name, years, len(a))) + for lo, hi, lab in bins: + sub = a[(a['gap'] > lo) & (a['gap'] <= hi)] + if len(sub) < 30: + continue + intra_up = (sub['intra'] > 0).mean() * 100 + lines.append("- {}（{}天，占比 {:.1f}%）: 日内均值 {}，日内收涨概率 {:.1f}%".format( + lab, len(sub), len(sub) / len(a) * 100, fmt_pct(sub['intra'].mean()), intra_up)) + lines.append("") + return chr(10).join(lines) + + +def weekday_effect(daily): + lines = ["## A股指数周内效应", "", + "> 口径：指数日收益按星期聚合（2003/2004-2026）。", ""] + wd_names = ['周一', '周二', '周三', '周四', '周五'] + for idx_code in ['sh000001', 'sh000300']: + a = daily[daily['ts_code'] == idx_code].dropna(subset=['ret']) + name = IDX_NAMES.get(idx_code, idx_code) + lines.append("### {}（{}天）".format(name, len(a))) + for wd in range(5): + sub = a[a['weekday'] == wd] + lines.append("- {}: {}".format(wd_names[wd], stat_line(sub['ret']))) + lines.append("") + return chr(10).join(lines) + + +def volatility_regime(daily): + lines = ["## 波动率机制切换（20日波动率翻倍/减半 -> 后5日）", "", + "> 口径：20日收益标准差 vs 前20日对比；统计切换后 5 日收益分布。", ""] + for idx_code in ['sh000001', 'sh000300']: + a = daily[daily['ts_code'] == idx_code].copy() + a = a.dropna(subset=['ret20', 'ret20_prev']) + name = IDX_NAMES.get(idx_code, idx_code) + lines.append("### {}".format(name)) + for lab, cond in [('骤增(>2倍)', a['ret20'] > a['ret20_prev'] * 2), + ('骤降(<0.5倍)', a['ret20'] < a['ret20_prev'] * 0.5)]: + sub_idx = a[cond].index + if len(sub_idx) < 30: + lines.append("- {}: 样本不足（{}）".format(lab, len(sub_idx))) + continue + fwd = [] + for idx in sub_idx: + pos = a.index.get_loc(idx) + window = a.iloc[pos + 1:pos + 6] + fwd.append((1 + window['ret']).prod() - 1) + fwd = pd.Series(fwd).dropna() + lines.append("- {}（{}次）: 后5日 {}".format(lab, len(sub_idx), stat_line(fwd))) + lines.append("") + return chr(10).join(lines) + + +def lhb_stats(lhb, daily): + lines = ["## 龙虎榜统计（近90个交易日）", "", + "> 口径：仅覆盖近90日采集窗口；上榜股只有当日快照，暂无次日数据（不提供次日胜率）。", ""] + total_up = (lhb['pct_change'] > 0).sum() + lines.append("- 上榜记录 {} 条；上榜当日上涨占比 {:.1f}%".format(len(lhb), total_up / len(lhb) * 100)) + if 'reason' in lhb.columns: + lines.append("- 下跌偏离上榜占比 {:.1f}%（跌幅榜多=弱势环境信号）".format( + lhb['reason'].str.contains('跌幅').mean() * 100)) + lines.append("") + lines.append("### 上榜原因分布 TOP10") + for reason, cnt in lhb['reason'].value_counts().head(10).items(): + lines.append("- {}次 | {}".format(cnt, reason)) + lines.append("") + daily_sh = daily[daily['ts_code'] == 'sh000001'][['trade_date', 'ret']] + merged = lhb.copy() + merged['trade_date'] = pd.to_datetime(merged['trade_date']) + m = merged.merge(daily_sh, on='trade_date', how='left') + m['mkt_down'] = m['ret'] < 0 + lines.append("### 大盘状态 × 龙虎榜家数") + for down, cnt in m.groupby('mkt_down').size().items(): + label = '指数下跌日' if down else '指数上涨日' + sub = m[m['mkt_down'] == down] + lines.append("- {}: 平均上榜 {:.1f} 条（{}条/{}天）".format(label, cnt / len(sub), cnt, len(sub))) + return chr(10).join(lines) + + +def main(): + import os + daily, g, lhb = load() + daily = enrich(daily) + refs = { + 'overnight_transmission.md': overnight_matrix(daily, g), + 'gap_intraday.md': gap_intraday(daily), + 'weekday_volatility.md': weekday_effect(daily) + NL2 + volatility_regime(daily), + 'lhb_stats.md': lhb_stats(lhb, daily), + } + os.makedirs(OUT, exist_ok=True) + for name, content in refs.items(): + out_path = os.path.normpath(os.path.join(OUT, name)) + with open(out_path, 'w', encoding='utf-8') as f: + f.write(content) + print("written:", name, len(content), "chars") + + +NL2 = chr(10) * 2 + +if __name__ == '__main__': + main() diff --git a/src/fetcher/__init__.py b/src/fetcher/__init__.py new file mode 100644 index 0000000..5924374 --- /dev/null +++ b/src/fetcher/__init__.py @@ -0,0 +1 @@ +# fetcher package diff --git a/src/fetcher/calendar_fetcher.py b/src/fetcher/calendar_fetcher.py new file mode 100644 index 0000000..1ff214a --- /dev/null +++ b/src/fetcher/calendar_fetcher.py @@ -0,0 +1,27 @@ +"""交易日历采集 - AkShare tool_trade_date_hist_sina""" +import warnings, logging, sqlite3 +from datetime import datetime +import pandas as pd +import akshare as ak + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +log = logging.getLogger(__name__) + +DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db' + +def fetch_calendar(): + df = ak.tool_trade_date_hist_sina() + log.info(f'获取到 {len(df)} 行，列: {df.columns.tolist()}') + # 原始只有 trade_date 一列；按 A 股规则补 is_trade_day + # 周六=0, 周日=0, 其他=1 + df['trade_date'] = pd.to_datetime(df['trade_date']) + df['is_trade_day'] = df['trade_date'].dt.dayofweek.apply(lambda x: 0 if x >= 5 else 1) + df['trade_date'] = df['trade_date'].dt.strftime('%Y-%m-%d') + conn = sqlite3.connect(DB) + df.to_sql('trade_calendar', conn, if_exists='replace', index=False) + conn.close() + log.info(f'写入 trade_calendar: {len(df)} 行') + return df + +if __name__ == '__main__': + fetch_calendar() diff --git a/src/fetcher/cn_news_fetcher.py b/src/fetcher/cn_news_fetcher.py new file mode 100644 index 0000000..60c4a6e --- /dev/null +++ b/src/fetcher/cn_news_fetcher.py @@ -0,0 +1,42 @@ +"""A股新闻采集 - AkShare stock_news_em""" +import warnings, logging, sqlite3 +from datetime import datetime +import pandas as pd +import akshare as ak + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +log = logging.getLogger(__name__) + +DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db' + +def fetch_news(): + df = ak.stock_news_em(symbol='A股') + log.info(f'获取到 {len(df)} 行，列: {df.columns.tolist()}') + # 列名标准化 + cn_map = { + '关键词': 'keywords', + '股票代码': 'ts_code', + '新闻标题': 'title', + '新闻内容': 'content', + '发布时间': 'pub_date', + '文章来源': 'source', + '新闻链接': 'url', + } + df = df.rename(columns=cn_map) + # 只保留表里有的列 + db_cols = ['title', 'content', 'pub_date', 'source', 'url', 'keywords'] + for c in db_cols: + if c not in df.columns: + df[c] = '' + df = df[['id'] + [c for c in db_cols if c in df.columns]] if 'id' in df.columns else df[db_cols] + df['inserted_at'] = datetime.now().isoformat() + if 'id' not in df.columns: + df.insert(0, 'id', range(1, len(df) + 1)) + conn = sqlite3.connect(DB) + df.to_sql('news_cn', conn, if_exists='replace', index=False) + conn.close() + log.info(f'写入 news_cn: {len(df)} 行') + return df + +if __name__ == '__main__': + fetch_news() diff --git a/src/fetcher/fund_flow_fetcher.py b/src/fetcher/fund_flow_fetcher.py new file mode 100644 index 0000000..d08ba49 --- /dev/null +++ b/src/fetcher/fund_flow_fetcher.py @@ -0,0 +1,151 @@ +"""资金流向采集 - 修正版：同花顺个股资金流（逐页校验，跳过异常页），历史保留""" +import time +import warnings +import logging +from urllib.parse import urlparse + +import pandas as pd +import requests + +from src.storage.db import upsert_rows + +warnings.filterwarnings('ignore') +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +log = logging.getLogger(__name__) + +DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db' + +# 出站请求域名白名单（SSRF 防护：仅 http + 白名单主机，禁重定向跟随） +_ALLOWED_HOSTS = {'data.10jqka.com.cn'} +_ALLOWED_PREFIX = 'http://data.10jqka.com.cn/funds/ggzjl/' + + +def _safe_get(url: str, headers: dict, timeout: int = 15) -> requests.Response: + """白名单校验后的 GET：协议/host 白名单 + 禁重定向，防 SSRF""" + u = urlparse(url) + if u.scheme != 'http' or u.hostname not in _ALLOWED_HOSTS: + raise ValueError(f'blocked non-allowlist url: {url}') + return requests.get(url, timeout=timeout, allow_redirects=False, headers=headers) + + +def parse_cn_amount(v): + """'1.5亿' -> 1.5e8, '-300万' -> -3e6，逗号/空白容错；无效返回 None""" + if v is None or (isinstance(v, float) and v != v): + return None + s = str(v).replace(',', '').replace(' ', '').strip() + if not s or s in ('--', '-', 'nan'): + return None + mult = 1.0 + if s.endswith('亿'): + mult, s = 1e8, s[:-1] + elif s.endswith('万'): + mult, s = 1e4, s[:-1] + try: + return float(s) * mult + except ValueError: + return None + + +def _ths_headers(): + from akshare.stock_feature.stock_fund_flow import _get_file_content_ths + from py_mini_racer import MiniRacer + js_code = MiniRacer() + js_code.eval(_get_file_content_ths("ths.js")) + v_code = js_code.call("v") + return { + "Accept": "text/html, */*; q=0.01", + "hexin-v": v_code, + "Host": "data.10jqka.com.cn", + "Pragma": "no-cache", + "Referer": "http://data.10jqka.com.cn/funds/hyzjl/", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/90.0.4430.85 Safari/537.36", + "X-Requested-With": "XMLHttpRequest", + } + + +def _page_url(page: int) -> str: + """拼分页 URL；page 强制 int，且必须落在白名单前缀内""" + page = int(page) + url = _ALLOWED_PREFIX + f"field/zdf/order/desc/page/{page}/ajax/1/free/1/" + if not url.startswith(_ALLOWED_PREFIX): + raise ValueError('url escapes allowlist prefix') + return url + + +def fetch_individual_flow(max_pages=None, retries_per_page=2): + """ + 同花顺个股资金流排行（修正版）： + akshare 原实现对部分页的异常 13 列表格直接崩溃，这里逐页校验： + 只接受恰好 10 列且表头匹配的表格，异常页跳过并记录。 + """ + from bs4 import BeautifulSoup + from io import StringIO + + headers = _ths_headers() # 每次请求刷新 hexin-v + expect = ['序号', '股票代码', '股票简称', '最新价', '涨跌幅', '换手率', + '流入资金(元)', '流出资金(元)', '净额(元)', '成交额(元)'] + + r = _safe_get(_page_url(1), headers) + soup = BeautifulSoup(r.text, features="lxml") + page_info = soup.find(name="span", attrs={"class": "page_info"}) + page_num = int(page_info.text.split("/")[1]) if page_info else 1 + if max_pages: + page_num = min(page_num, max_pages) + + frames = [] + skipped = [] + for page in range(1, page_num + 1): + for attempt in range(retries_per_page): + try: + r = _safe_get(_page_url(page), headers) + dfs = pd.read_html(StringIO(r.text)) + df = next((d for d in dfs if len(d.columns) == 10 + and list(d.columns)[:2] == ['序号', '股票代码']), None) + if df is None: + skipped.append(page) + break + df.columns = expect + frames.append(df) + break + except Exception: + if attempt == retries_per_page - 1: + skipped.append(page) + else: + time.sleep(1) + time.sleep(0.3) # 限速，避免同花顺封禁 + if skipped: + log.warning(f'跳过异常页: {skipped}') + if not frames: + return pd.DataFrame() + big = pd.concat(frames, ignore_index=True) + big.columns = [c.split('(')[0] for c in big.columns] # 去掉 (元) 后缀 + return big + + +def fetch_money_flow(): + """采集全市场个股当日主力净流入（净额），并入库（历史保留）""" + today = time.strftime('%Y-%m-%d') + raw = fetch_individual_flow() + if raw.empty: + log.warning('个股资金流采集为空') + return None + + df = pd.DataFrame({ + 'ts_code': raw['股票代码'].astype(str).str.zfill(6), + 'trade_date': today, + 'main_net_in': raw['净额'].map(parse_cn_amount), + 'in_flow': raw['流入资金'].map(parse_cn_amount), + 'out_flow': raw['流出资金'].map(parse_cn_amount), + 'amount': raw['成交额'].map(parse_cn_amount), + 'pct_change': raw['涨跌幅'].str.rstrip('%').map( + lambda x: float(x) if x and x not in ('--', '') else None), + }) + df = df[df['main_net_in'].notna()] + n = upsert_rows(df, 'money_flow', conflict_cols=['ts_code', 'trade_date']) + log.info(f'写入 money_flow: {n} 行（历史保留）') + return df + + +if __name__ == '__main__': + fetch_money_flow() diff --git a/src/fetcher/global_index_fetcher.py b/src/fetcher/global_index_fetcher.py new file mode 100644 index 0000000..c7b8d1c --- /dev/null +++ b/src/fetcher/global_index_fetcher.py @@ -0,0 +1,54 @@ +"""外盘指数采集 - AkShare index_us_stock_sina""" +import warnings, logging, sqlite3 +from datetime import datetime +import pandas as pd +import akshare as ak + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +log = logging.getLogger(__name__) + +DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db' + +# 要采集的外盘指数 +INDICES = [ + ('US.DJI', '.DJI', '道琼斯工业'), + ('US.NDX', '.NDX', '纳斯达克100'), + ('US.SPX', '.INX', '标普500'), +] + +def fetch_global_index(): + conn = sqlite3.connect(DB) + results = [] + for code, symbol, name in INDICES: + try: + df = ak.index_us_stock_sina(symbol=symbol) + log.info(f'{name}({symbol}): {len(df)} 行，列: {df.columns.tolist()}') + # 标准化 + cn_map = { + '时间': 'trade_date', + '日期': 'trade_date', + 'date': 'trade_date', + '收盘': 'close', + 'close': 'close', + '涨跌幅': 'pct_change', + 'pct_change': 'pct_change', + } + df = df.rename(columns=cn_map) + if 'trade_date' not in df.columns: + df['trade_date'] = df['date'] if 'date' in df.columns else None + df['index_code'] = code + df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y-%m-%d') + # 计算涨跌幅（基于前一日收盘价） + df = df.sort_values('trade_date') + df['pct_change'] = df['close'].pct_change() * 100 + out = df[['index_code', 'trade_date', 'close', 'pct_change']].dropna() + out.to_sql('global_index', conn, if_exists='append', index=False) + results.append((code, len(out))) + log.info(f' -> 写入 {code}: {len(out)} 行') + except Exception as e: + log.warning(f'{name}({symbol}) 失败: {e}') + conn.close() + return results + +if __name__ == '__main__': + fetch_global_index() diff --git a/src/fetcher/lhb_fetcher.py b/src/fetcher/lhb_fetcher.py new file mode 100644 index 0000000..b4473c6 --- /dev/null +++ b/src/fetcher/lhb_fetcher.py @@ -0,0 +1,83 @@ +""" +龙虎榜采集器 +主方案：AkShare stock_lhb_detail_daily_sina（已实测可用，含上榜原因） +备选：东财 datacenter 接口（报表名不确定，暂时注释备用） +""" + +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +import akshare as ak +import pandas as pd +from datetime import datetime, timedelta +from src.storage.db import upsert_df, upsert_log, get_last_fetch + + +def fetch_lhb_date(trade_date: str) -> pd.DataFrame: + """ + 获取单日龙虎榜 + trade_date: "YYYY-MM-DD" 或 "YYYYMMDD" + """ + # 统一为 YYYYMMDD 格式 + ds = trade_date.replace("-", "") + try: + df = ak.stock_lhb_detail_daily_sina(date=ds) + except Exception as e: + print(f" [WARN] LHB Sina failed for {trade_date}: {e}") + return pd.DataFrame() + + if df is None or df.empty: + return pd.DataFrame() + + # Sina 返回列（实测）：序号, 股票代码, 股票名称, 收盘价, 涨跌幅, 成交额, 成交额, 上榜原因 + # 编码问题导致中文列名显示乱码，用位置索引映射 + # 0=序号, 1=代码, 2=名称, 3=收盘, 4=涨跌幅, 5=成交额, 6=??? 7=上榜原因 + df.columns = ["no", "ts_code", "name", "close", "pct_change", "amount", "amount2", "reason"] + + df["trade_date"] = trade_date.replace("-", "") + # 格式化为 YYYY-MM-DD + df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d") + + # 保留有用字段 + keep = ["trade_date", "ts_code", "name", "close", "pct_change", "amount", "reason"] + df = df[[c for c in keep if c in df.columns]] + return df + + +def fetch_lhb_range(start_date: str, end_date: str = None) -> pd.DataFrame: + """抓取日期区间的龙虎榜（逐日）""" + end_date = end_date or datetime.today().strftime("%Y-%m-%d") + all_days = [] + cur = datetime.strptime(start_date.replace("-", ""), "%Y%m%d") + end = datetime.strptime(end_date.replace("-", ""), "%Y%m%d") + while cur <= end: + ds = cur.strftime("%Y-%m-%d") + df = fetch_lhb_date(ds) + if df is not None and not df.empty: + all_days.append(df) + cur += timedelta(days=1) + if all_days: + return pd.concat(all_days, ignore_index=True) + return pd.DataFrame() + + +def incremental_lhb(source: str = "lhb_daily") -> pd.DataFrame: + """增量采集：只抓上次之后的新交易日""" + last = get_last_fetch(source) + start = last.get("last_date") or (datetime.today() - timedelta(days=7)).strftime("%Y-%m-%d") + end = datetime.today().strftime("%Y-%m-%d") + df = fetch_lhb_range(start, end) + if df is not None and not df.empty: + upsert_df(df, "lhb_daily") + upsert_log(source, param="", last_date=end) + return df + + +if __name__ == "__main__": + import fire + fire.Fire({ + "date": lambda d: upsert_df(fetch_lhb_date(d), "lhb_daily"), + "range": lambda s, e=None: upsert_df(fetch_lhb_range(s, e), "lhb_daily"), + "incr": lambda: incremental_lhb(), + }) diff --git a/src/fetcher/macro_event_fetcher.py b/src/fetcher/macro_event_fetcher.py new file mode 100644 index 0000000..9ca817b --- /dev/null +++ b/src/fetcher/macro_event_fetcher.py @@ -0,0 +1,133 @@ +""" +每日大事提醒 / 财经日历采集器 +主方案：AkShare news_economic_baidu（财经日历事件，日期/时间/重要性完整） +备选：东财 datacenter 大事提醒接口 +""" + +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +import requests +import pandas as pd +import akshare as ak +from datetime import datetime, timedelta +from src.storage.db import upsert_df, upsert_rows, upsert_log, get_last_fetch + +EM_BASE = "https://datacenter-web.eastmoney.com/api/data/v1/get" + + +def em_headers() -> dict: + return { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Referer": "https://data.eastmoney.com/", + "Accept": "application/json", + } + + +# ── 主方案：AkShare 财经日历 ───────────────────────────────────────────────── + +def fetch_macro_calendar() -> pd.DataFrame: + """ + AkShare news_economic_baidu 返回财经日历事件 + 列: 标题, 时间, 重要性, 前值, 预测值, 公布值, 影响 + """ + try: + df = ak.news_economic_baidu() + except Exception as e: + print(f" [WARN] news_economic_baidu failed: {e}") + return pd.DataFrame() + + if df is None or df.empty: + return pd.DataFrame() + + # 实测列（akshare 1.18.92）: 日期, 时间, 地区, 事件, 公布, 预期, 前值, 重要性 + # 按列【名称】映射，避免位置错位（严禁再按位置覆盖 df.columns） + col_map = {'日期': 'event_date', '时间': 'event_time', '地区': 'source', + '事件': 'title', '公布': 'actual', '预期': 'forecast', + '前值': 'prev', '重要性': 'event_type'} + df = df.rename(columns=col_map) + + # 防御：akshare 偶发返回缺列，补默认值 + for c, default in [('event_time', ''), ('actual', None), ('forecast', None), + ('prev', None), ('event_type', 0), ('source', '')]: + if c not in df.columns: + df[c] = default + df["title"] = df["title"].fillna("").astype(str) + df["event_date"] = pd.to_datetime(df["event_date"], errors="coerce").dt.strftime("%Y-%m-%d") + df["event_time"] = df["event_time"].fillna("").astype(str) + df["event_type"] = pd.to_numeric(df["event_type"], errors="coerce").fillna(0).astype(int) + df["source"] = df["source"].fillna("").astype(str) + + keep = ["event_date", "event_time", "title", "event_type", "source", + "actual", "forecast", "prev"] + df = df[[c for c in keep if c in df.columns]] + return df + + +# ── 备选：东财大事提醒接口 ─────────────────────────────────────────────────── + +def fetch_em_events( + start_date: str = None, + end_date: str = None, + page: int = 1, + page_size: int = 50 +) -> pd.DataFrame: + """ + 东财大事提醒接口（需确认报表名） + 已知可用报表: RPT_MAJOR_NOTICE（公告）、RPT_LHB_ALL_STOCKS（龙虎榜） + 大事提醒报表名待抓包确认，暂时返回空 + """ + if start_date is None: + start_date = (datetime.today() - timedelta(days=3)).strftime("%Y-%m-%d") + if end_date is None: + end_date = (datetime.today() + timedelta(days=60)).strftime("%Y-%m-%d") + + # 尝试 RPT_MAJOR_NOTICES（公告大事） + report_names = ["RPT_MAJOR_NOTICES", "RPT_IMPORTANT_NEWS"] + for report in report_names: + params = { + "reportName": report, + "columns": "ALL", + "filter": f'(EVENT_DATE>=\'{start_date}\') and (EVENT_DATE<=\'{end_date}\')', + "pageNumber": page, + "pageSize": page_size, + "sortTypes": "1", + "sortColumns": "EVENT_DATE", + "source": "WEB", + "client": "WEB", + } + try: + r = requests.get(EM_BASE, params=params, headers=em_headers(), timeout=15) + r.raise_for_status() + resp = r.json() + result = resp.get("result", {}) or {} + data = result.get("data", []) or [] + if data: + df = pd.DataFrame(data) + df.columns = [c.lower() for c in df.columns] + return df + except Exception: + continue + return pd.DataFrame() + + +def incremental_events(source: str = "macro_event") -> pd.DataFrame: + """增量采集宏观事件（按 event_date+event_time+title 去重，历史保留）""" + df = fetch_macro_calendar() + if df is not None and not df.empty: + need = [c for c in ['event_date', 'event_time', 'title', 'event_type', 'source', + 'actual', 'forecast', 'prev', 'inserted_at'] if c in df.columns] + upsert_rows(df[need], source, + conflict_cols=['event_date', 'event_time', 'title']) + upsert_log(source, "", last_date=datetime.today().strftime("%Y-%m-%d")) + return df + + +if __name__ == "__main__": + import fire + fire.Fire({ + "ak": lambda: upsert_df(fetch_macro_calendar(), "macro_event"), + "em": lambda s=None, e=None: upsert_df(fetch_em_events(s, e), "macro_event"), + "incr": lambda: incremental_events(), + }) diff --git a/src/fetcher/news_flash_fetcher.py b/src/fetcher/news_flash_fetcher.py new file mode 100644 index 0000000..f87f9d0 --- /dev/null +++ b/src/fetcher/news_flash_fetcher.py @@ -0,0 +1,141 @@ +""" +7x24 快讯流采集器 +主方案：东财 np-anotice-stock 快讯接口（秒级时间戳） +备选方案：AkShare news_economic_baidu（财经日历事件） +""" + +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +import requests +import pandas as pd +import akshare as ak +from datetime import datetime +from src.storage.db import upsert_df, upsert_log, get_last_fetch + + +# ── 主方案：东财快讯接口 ──────────────────────────────────────────────────── + +EM_FLASH_API = "https://np-anotice-stock.eastmoney.com/api/security/ann" + +def fetch_em_flash(category: str = "全部", page: int = 1, page_size: int = 50) -> dict | None: + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", + "Referer": "https://www.eastmoney.com/", + "Accept": "application/json", + } + params = { + "sr": -1, + "page": page, + "pageSize": page_size, + "category": category, + "type": "", + } + try: + r = requests.get(EM_FLASH_API, params=params, headers=headers, timeout=15) + r.raise_for_status() + return r.json() + except Exception as e: + print(f" [WARN] EastMoney flash request failed: {e}") + return None + + +def parse_em_flash(resp: dict) -> list[dict]: + """解析东财快讯响应""" + try: + items = resp.get("data", []) or resp.get("list", []) or [] + return items + except Exception: + return [] + + +def fetch_flash_pages(category: str = "全部", pages: int = 3, page_size: int = 30) -> pd.DataFrame: + """抓取多页快讯""" + all_items = [] + for page in range(1, pages + 1): + resp = fetch_em_flash(category, page=page, page_size=page_size) + if resp is None: + break + items = parse_em_flash(resp) + if not items: + break + all_items.extend(items) + + if not all_items: + return pd.DataFrame() + + df = pd.DataFrame(all_items) + df.columns = [c.lower() for c in df.columns] + + # 字段映射（东财字段 → 标准名） + col_map = { + "title": "content", + "showtime": "event_time", + "time": "event_time", + "notice_date":"event_time", + "source": "source", + "media": "source", + "site": "source", + "url": "url", + } + df = df.rename(columns={k: v for k, v in col_map.items() if k in df.columns}) + + # 时间标准化 + if "event_time" in df.columns: + df["event_time"] = pd.to_datetime(df["event_time"], errors="coerce").dt.strftime("%Y-%m-%d %H:%M:%S") + + keep = ["content", "event_time", "source"] + df = df[[c for c in keep if c in df.columns]] + return df + + +# ── 备选方案：AkShare 新闻 ─────────────────────────────────────────────────── + +def fetch_akshare_news() -> pd.DataFrame: + """AkShare 东财财经新闻（带发布时间戳） + stock_news_em 返回列: [关键词, 股票代码, 新闻标题, 发布时间, 新闻来源, 新闻内容] + """ + try: + df = ak.stock_news_em(symbol="A股") + except Exception as e: + print(f" [WARN] AkShare news failed: {e}") + return pd.DataFrame() + + if df is None or df.empty: + return pd.DataFrame() + + # AkShare 返回中文列名，用位置映射更稳定 + # 列0=关键词, 1=股票代码, 2=新闻标题(content), 3=发布时间, 4=新闻来源, 5=新闻内容 + df.columns = ["keyword", "ts_code", "content", "event_time", "source", "body"] + if "event_time" in df.columns: + df["event_time"] = pd.to_datetime(df["event_time"], errors="coerce").dt.strftime("%Y-%m-%d %H:%M:%S") + + keep = ["content", "event_time", "source"] + return df[keep] + + +# ── 入口函数 ───────────────────────────────────────────────────────────────── + +def incremental_flash(source: str = "news_flash") -> pd.DataFrame: + """ + 增量采集：优先东财快讯，降级到 AkShare 新闻 + """ + # 尝试东财快讯（category="全部" 获取最广） + df = fetch_flash_pages(category="全部", pages=3) + if df is None or df.empty: + df = fetch_akshare_news() + + if df is not None and not df.empty: + upsert_df(df, source, csv备份=False) + upsert_log(source, "", last_ts=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) + return df + + +if __name__ == "__main__": + import fire + fire.Fire({ + "em": lambda c="全部", p=3: upsert_df(fetch_flash_pages(c, pages=p), "news_flash"), + "ak": lambda: upsert_df(fetch_akshare_news(), "news_flash"), + "incr": lambda: incremental_flash(), + }) diff --git a/src/fetcher/stock_api.py b/src/fetcher/stock_api.py new file mode 100644 index 0000000..97e555f --- /dev/null +++ b/src/fetcher/stock_api.py @@ -0,0 +1,179 @@ +""" +日K线采集器 — AkShare（免费，无需 Token） +支持前复权日线、分钟线（东财） +""" + +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +import akshare as ak +import pandas as pd +from datetime import datetime, timedelta +from src.storage.db import upsert_df, upsert_rows, upsert_log, get_last_fetch + + +def fetch_daily( + ts_code: str, + start_date: str = None, + end_date: str = None, + adjust: str = "qfq" +) -> pd.DataFrame | None: + """ + 获取单只A股日K线（前复权） + ts_code: 6位代码，如 "000001" + start_date / end_date: "YYYYMMDD" + adjust: "qfq" 前复权 / "hfq" 后复权 / "" 不复权 + """ + try: + df = ak.stock_zh_a_hist( + symbol=ts_code, + period="daily", + start_date=start_date or (datetime.today() - timedelta(days=365)).strftime("%Y%m%d"), + end_date=end_date or datetime.today().strftime("%Y%m%d"), + adjust=adjust + ) + except Exception as e: + print(f" [WARN] stock_zh_a_hist failed for {ts_code}: {e}") + return None + + if df is None or df.empty: + return None + + # AkShare 列名统一转小写 + df.columns = [c.lower() for c in df.columns] + + # 统一列名映射（兼容多种 AkShare 版本列名） + col_map = {} + for old in df.columns: + ol = old.lower() + if ol in ("日期", "date"): + col_map[old] = "trade_date" + elif ol in ("股票代码", "代码", "symbol", "code"): + col_map[old] = "ts_code" + elif ol in ("开盘", "open"): + col_map[old] = "open" + elif ol in ("最高", "high"): + col_map[old] = "high" + elif ol in ("最低", "low"): + col_map[old] = "low" + elif ol in ("收盘", "close"): + col_map[old] = "close" + elif ol in ("成交量", "volume", "vol"): + col_map[old] = "volume" + elif ol in ("成交额", "amount", "amt"): + col_map[old] = "amount" + df = df.rename(columns=col_map) + + # 确保 ts_code 列存在 + if "ts_code" not in df.columns: + df["ts_code"] = ts_code + + # 只保留核心列 + keep = ["ts_code", "trade_date", "open", "high", "low", "close", "volume", "amount"] + df = df[[c for c in keep if c in df.columns]] + + # 日期格式统一为 YYYY-MM-DD + if "trade_date" in df.columns: + df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d") + + return df + + +def fetch_index_daily( + symbol: str = "sh000001", # 注意：AkShare 需 sh/zs 前缀，如 sh000001 + start_date: str = None, + end_date: str = None +) -> pd.DataFrame | None: + """获取指数日K线（用于大盘背景板）""" + try: + df = ak.stock_zh_index_daily(symbol=symbol) + except Exception as e: + print(f" [WARN] stock_zh_index_daily failed for {symbol}: {e}") + return None + + if df is None or df.empty: + return None + df.columns = [c.lower() for c in df.columns] + + # AkShare 1.18+ 返回列名: date, open, high, low, close, volume（无 amount） + col_map = {"date": "trade_date", "open": "open", "high": "high", + "low": "low", "close": "close", "volume": "volume"} + df = df.rename(columns={k: v for k, v in col_map.items() if k in df.columns}) + # amount 列不存在于指数，设为 NaN 保持 schema 一致 + if "amount" not in df.columns: + df["amount"] = None + + if "trade_date" in df.columns: + df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d") + + # 按日期过滤 + if start_date: + df = df[df["trade_date"] >= start_date] + if end_date: + df = df[df["trade_date"] <= end_date] + + if "ts_code" not in df.columns: + df["ts_code"] = symbol + return df + + +def batch_fetch_indices( + symbols: list[str] = None, + days: int = 90 +) -> pd.DataFrame: + """批量获取主要指数（沪深300/上证/创业板等）""" + # AkShare 前缀：sh=上证, sz=深圳 + symbols = symbols or [ + "sh000001", # 上证指数 + "sz399001", # 深证成指 + "sz399006", # 创业板指 + "sh000300", # 沪深300 + "sh000016", # 上证50 + ] + end = datetime.today().strftime("%Y-%m-%d") + start = (datetime.today() - timedelta(days=days)).strftime("%Y-%m-%d") + + results = [] + for sym in symbols: + df = fetch_index_daily(sym) + if df is not None and not df.empty: + # 日期过滤在内存做 + if "trade_date" in df.columns: + df = df[(df["trade_date"] >= start) & (df["trade_date"] <= end)] + if not df.empty: + results.append(df) + + if results: + return pd.concat(results, ignore_index=True) + return pd.DataFrame() + + +def run(ts_codes: list[str] = None, days: int = 90): + """ + 采集入口 + ts_codes: 指定代码列表，None 时采集主要指数 + """ + end = datetime.today().strftime("%Y%m%d") + start = (datetime.today() - timedelta(days=days)).strftime("%Y%m%d") + + if ts_codes: + for code in ts_codes: + df = fetch_daily(code, start, end) + if df is not None and not df.empty: + upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"]) + upsert_log("akshare_daily", code, last_date=end) + else: + # 默认采集主要指数 + df = batch_fetch_indices(days=days) + if df is not None and not df.empty: + upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"]) + + +if __name__ == "__main__": + import fire + fire.Fire({ + "daily": lambda ts_code: upsert_df(fetch_daily(ts_code), "stock_daily"), + "indices": lambda: run(), + "batch": lambda ts_codes: run(ts_codes=ts_codes), + }) diff --git a/src/pipeline/collect_all.py b/src/pipeline/collect_all.py new file mode 100644 index 0000000..b1ccaa7 --- /dev/null +++ b/src/pipeline/collect_all.py @@ -0,0 +1,48 @@ +""" +P0 全量采集脚本 — 依次执行所有 P0 数据源 +用法: python -m src.pipeline.collect_all +""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from datetime import datetime, timedelta + +def run(): + print(f"\n{'='*50}") + print(f" A股时间线数据采集 — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print(f"{'='*50}\n") + + # 1. 日K线（主要指数） + print("[1/4] 日K线（主要指数 90天）...") + from src.fetcher import stock_api + stock_api.run() + + # 2. 龙虎榜（近7日） + print("\n[2/4] 龙虎榜（近7日）...") + from src.fetcher import lhb_fetcher + start = (datetime.today() - timedelta(days=7)).strftime("%Y-%m-%d") + end = datetime.today().strftime("%Y-%m-%d") + df = lhb_fetcher.fetch_lhb_range(start, end) + from src.storage.db import upsert_df, upsert_log + if df is not None and not df.empty: + upsert_df(df, "lhb_daily") + upsert_log("lhb_daily", "", last_date=end) + + # 3. 7x24 快讯（最新3页） + print("\n[3/4] 7x24 快讯流...") + from src.fetcher import news_flash_fetcher + news_flash_fetcher.incremental_flash() + + # 4. 大事提醒 + print("\n[4/4] 每日大事提醒...") + from src.fetcher import macro_event_fetcher + macro_event_fetcher.incremental_events() + + print(f"\n{'='*50}") + print(f" 采集完成") + print(f"{'='*50}\n") + + +if __name__ == "__main__": + run() diff --git a/src/realtime/realtime_analyzer.py b/src/realtime/realtime_analyzer.py new file mode 100644 index 0000000..91cfd41 --- /dev/null +++ b/src/realtime/realtime_analyzer.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +""" +实时市场时间线分析器（常驻运行，不自动退出） +========================================== +功能（每分钟一轮，按市场状态自动分派）： + 1. 盘前（09:15 前）：美股隔夜收盘 → 依据 a-stock-timeline-patterns 实证先验 + 输出今日 A 股跳空/日内路径的历史概率预估 + 2. 盘中（09:15-15:05）：上证指数实时点位相对昨收位置 + 快讯关键词告警 + 3. 盘后（15:10）：当日龙虎榜入库 + 日报（指数涨跌/跳空/日内分解） +输出：控制台 + 追加写入 ~/.a_stock_timeline/realtime_feed.jsonl 与 signal 日志 +停止：Ctrl+C（优雅退出） +""" +import io +import json +import os +import sqlite3 +import sys +import time +import traceback +from datetime import datetime, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import warnings +warnings.filterwarnings('ignore') + +import akshare as ak +import requests + +ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +DB = os.path.join(ROOT, 'data', 'a_stock.db') +FEED = os.path.join(os.path.expanduser('~'), '.a_stock_timeline', 'realtime_feed.jsonl') +SKILL_REF = r'E:\Data\skills\a-stock-timeline-patterns\references\overnight_transmission.md' + +KEYWORDS = ['降息', '降准', '加息', '关税', '制裁', '收购', '重组', '国债', '证监会', 'PMI', 'CPI'] + +HIS = None # 历史先验缓存 {('NDX'|'SPX'|'DJI', bin): {...}} + + +def load_priors(): + """解析实证先验表（overnight_transmission.md），供盘前预估""" + global HIS + import re + HIS = {} + path = SKILL_REF + if not os.path.exists(path): + return + cur_bin, cur_us, cur_idx = None, None, None + with io.open(path, encoding='utf-8') as f: + for line in f: + m = re.match(r'### (\S+)T日 -> (\S+)T\+1日', line) + if m: + cur_us, cur_idx = m.group(1), m.group(2) + continue + m = re.match(r'- 美股(\S+?)（(\d+)天）: A股跳空 N=\d+, 均值 (-?[\d.]+)%, 中位数 (-?[\d.]+)%, 胜率 ([\d.]+)%; ' + r'日内 N=\d+, 均值 (-?[\d.]+)%, 中位数 (-?[\d.]+)%, 胜率 ([\d.]+)%', line) + if m and cur_us and cur_idx: + cur_bin = m.group(1) + HIS[(cur_us, cur_idx, cur_bin)] = { + 'n': int(m.group(2)), + 'gap_mean': float(m.group(3)), 'gap_median': float(m.group(4)), + 'gap_win': float(m.group(5)), + 'intra_mean': float(m.group(6)), 'intra_median': float(m.group(7)), + 'intra_win': float(m.group(8)), + } + print('[init] 实证先验加载: {} 条'.format(len(HIS)), flush=True) + + +def db_conn(): + conn = sqlite3.connect(DB, check_same_thread=False) + return conn + + +def us_overnight(conn): + """最近一个已收盘的美股交易日涨跌（三大指数）""" + out = {} + for code in ('US.NDX', 'US.SPX', 'US.DJI'): + row = conn.execute( + "SELECT trade_date, pct_change FROM global_index WHERE index_code=? " + "ORDER BY trade_date DESC LIMIT 1", (code,)).fetchone() + if row: + out[code] = {'date': row[0], 'pct': row[1] if row[1] is not None else 0.0} + return out + + +def forecast_from_priors(us): + """按先验表预估：取纳斯达克分箱对应的上证跳空/日内历史分布""" + if not HIS or 'US.NDX' not in us: + return None + pct = us['US.NDX']['pct'] / 100.0 + if pct <= -0.015: + b = '跌>1.5%' + elif pct <= -0.005: + b = '跌0.5~1.5%' + elif pct < 0.005: + b = '正负0.5%内' + elif pct < 0.015: + b = '涨0.5~1.5%' + else: + b = '涨>1.5%' + k = ('US.NDX', 'sh000001', b) + if k not in HIS: + return None + p = HIS[k] + return ('隔夜预估[美股纳指{:+.2f}% -> 分箱"{}"]: 历史上上证次日跳空均值 {:+.2f}%（低开概率 {:.0f}%），' + '日内均值 {:+.2f}%（日内收涨概率 {:.0f}%），全天均值 {:+.2f}%。' + '（样本{}天，美股99-24先验，仅参考）').format( + us['US.NDX']['pct'], b, p['gap_mean'], 100 - p['gap_win'], + p['intra_mean'], p['intra_win'], p['gap_mean'] + p['intra_mean'], p['n']) + + +def index_spot_sh(): + """上证指数实时点位（新浪，免token）""" + try: + df = ak.stock_zh_index_spot_em(symbol='上证系列指数') + row = df[df['名称'] == '上证指数'] + if not row.empty: + r = row.iloc[0] + return {'price': float(r['最新价']), 'pct': float(r['涨跌幅'])} + except Exception: + pass + try: + df = ak.stock_zh_index_spot_sina() + row = df[df['代码'] == 'sh000001'] + if not row.empty: + r = row.iloc[0] + return {'price': float(r['最新价']), 'pct': float(r['涨跌幅'])} + except Exception: + pass + return None + + +def today_lhb_count(conn): + today = datetime.now().strftime('%Y-%m-%d') + n = conn.execute("SELECT COUNT(*) FROM lhb_daily WHERE trade_date=?", (today,)).fetchone()[0] + return n + + +def news_alerts(conn): + """扫描快讯/新闻关键词命中（近2小时新增）""" + hits = [] + cutoff = (datetime.now() - timedelta(minutes=30)).strftime('%Y-%m-%d %H:%M:%S') + for table, tcol, ccol in (('news_flash', 'event_time', 'content'), + ('news_cn', 'pub_date', 'title')): + try: + rows = conn.execute( + f"SELECT {tcol}, {ccol} FROM {table} WHERE {tcol} >= ? ORDER BY {tcol} DESC LIMIT 50", + (cutoff,)).fetchall() + except sqlite3.OperationalError: + continue + for ts, text in rows: + for kw in KEYWORDS: + if kw in (text or ''): + hits.append({'time': ts, 'kw': kw, 'text': (text or '')[:80]}) + break + return hits + + +def emit(kind, payload): + rec = {'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'kind': kind, 'data': payload} + line = json.dumps(rec, ensure_ascii=False) + os.makedirs(os.path.dirname(FEED), exist_ok=True) + with io.open(FEED, 'a', encoding='utf-8') as f: + f.write(line + '\n') + print('[{}] {} {}'.format(rec['ts'], kind, json.dumps(payload, ensure_ascii=False)), flush=True) + + +def save_lhb_today(conn): + """收盘后采集当日龙虎榜""" + try: + from src.fetcher.lhb_fetcher import fetch_lhb # 项目内已有实现 + df = fetch_lhb() + if df is not None and not df.empty: + from src.storage.db import upsert_rows + upsert_rows(df, 'lhb_daily', conflict_cols=['trade_date', 'ts_code', 'reason']) + return len(df) + except Exception as e: + print('[lhb] 采集失败:', e, flush=True) + return 0 + + + + +def fetch_em_flash(conn): + """东财 7x24 快讯轮询：增量入库 news_flash（按 content+event_time 去重）""" + try: + r = requests.get( + 'https://np-weblist.eastmoney.com/comm/web/getFastNewsList', + params={'client': 'web', 'biz': 'web_724', 'fastColumn': '102', + 'sortEnd': '', 'pageSize': '20', 'req_trace': '1'}, + headers={'User-Agent': 'Mozilla/5.0'}, timeout=10) + data = r.json().get('data', {}) or {} + rows = [] + for n in data.get('fastNewsList', []) or []: + ts = n.get('showTime', '') + summary = (n.get('summary') or n.get('title') or '').strip() + if ts and summary: + dup = conn.execute( + "SELECT 1 FROM news_flash WHERE event_time=? AND content=? LIMIT 1", + (ts, summary)).fetchone() + if not dup: + conn.execute( + "INSERT INTO news_flash(content, event_time, source, inserted_at) " + "VALUES (?,?,?,datetime('now','localtime'))", + (summary, ts, '东财7x24')) + rows.append((ts, summary[:60])) + conn.commit() + return rows + except Exception as e: + print('[flash] 拉取失败:', e, flush=True) + return [] + +def one_cycle(conn, state): + now = datetime.now() + t = now.time() + hm = t.hour * 100 + t.minute + + # 1) 盘前（07:00-09:25）：隔夜预估 + if 700 <= hm < 925 and not state.get('premarket_done'): + us = us_overnight(conn) + fc = forecast_from_priors(us) + emit('盘前隔夜预估', {'us': us, 'forecast': fc}) + state['premarket_done'] = True + + # 2) 盘中（09:25-15:05）：实时点位 + 快讯告警 + if 925 <= hm < 1505: + for ts, txt in fetch_em_flash(conn): + emit('新快讯', {'time': ts, 'text': txt}) + spot = index_spot_sh() + if spot and spot.get('price', 0) > 0: + if 'session_open_price' not in state or hm < 940 and state.get('last_price') is None: + pass + emit('盘中点位', {'上证': spot['price'], '涨跌幅%': spot['pct']}) + hits = news_alerts(conn) + for h in hits: + emit('快讯关键词告警', h) + + # 3) 盘后（15:10 后）：龙虎榜 + 日报 + if hm >= 1510 and not state.get('postmarket_done'): + n = save_lhb_today(conn) + sh = conn.execute("SELECT trade_date, close FROM stock_daily WHERE ts_code='sh000001' " + "ORDER BY trade_date DESC LIMIT 1").fetchone() + emit('收盘日报', {'龙虎榜新增': n, '上证最新收盘': sh}) + state['postmarket_done'] = True + + # 4) 日切重置 + if state.get('date') != now.strftime('%Y-%m-%d'): + state.clear() + state['date'] = now.strftime('%Y-%m-%d') + load_priors() + emit('日切', {'date': state['date']}) + + +def main(): + print('=== JQuant 市场时间线实时分析器 ===') + print('数据库:', DB) + print('输出流:', FEED) + print('Ctrl+C 停止', flush=True) + conn = db_conn() + load_priors() + state = {'date': datetime.now().strftime('%Y-%m-%d')} + interval = 60 + while True: + try: + one_cycle(conn, state) + except KeyboardInterrupt: + print('收到停止指令，退出', flush=True) + break + except Exception: + traceback.print_exc() + time.sleep(10) + time.sleep(interval) + + +if __name__ == '__main__': + main() diff --git a/src/storage/__init__.py b/src/storage/__init__.py new file mode 100644 index 0000000..2dc7fd4 --- /dev/null +++ b/src/storage/__init__.py @@ -0,0 +1 @@ +# storage package diff --git a/src/storage/db.py b/src/storage/db.py new file mode 100644 index 0000000..8ade8a9 --- /dev/null +++ b/src/storage/db.py @@ -0,0 +1,250 @@ +""" +SQLite + CSV 双写存储层 +表结构按"接口族"隔离，方便按需降级 +""" + +import sqlite3 +from pathlib import Path +import pandas as pd +from datetime import datetime +from typing import Optional, Literal + +DB_PATH = Path(__file__).parent.parent.parent / "data" / "a_stock.db" +CSV_DIR = Path(__file__).parent.parent.parent / "data" / "raw" + +# ───────────────────────────────────────────────────────────────────────────── +# 连接管理器 +# ───────────────────────────────────────────────────────────────────────────── + +def get_conn() -> sqlite3.Connection: + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(DB_PATH, check_same_thread=False) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + +def to_csv(df: pd.DataFrame, table: str, ts: Optional[str] = None): + """入库后同步写一份 CSV 备份（方便溯源）""" + if df is None or df.empty: + return + ts = ts or datetime.now().strftime("%Y%m%d_%H%M%S") + CSV_DIR.mkdir(parents=True, exist_ok=True) + path = CSV_DIR / f"{table}_{ts}.csv" + df.to_csv(path, index=False, encoding="utf-8-sig") + return path + + +# ───────────────────────────────────────────────────────────────────────────── +# Schema 创建 +# ───────────────────────────────────────────────────────────────────────────── + +SCHEMA_SQL = """ +-- 交易日历 +CREATE TABLE IF NOT EXISTS trade_calendar ( + trade_date TEXT PRIMARY KEY, + is_trade_day INTEGER NOT NULL +); + +-- A股日K线（前复权） +CREATE TABLE IF NOT EXISTS stock_daily ( + ts_code TEXT NOT NULL, + trade_date TEXT NOT NULL, + open REAL, + high REAL, + low REAL, + close REAL, + volume REAL, + amount REAL, + PRIMARY KEY (ts_code, trade_date) +); + +-- 龙虎榜 +CREATE TABLE IF NOT EXISTS lhb_daily ( + trade_date TEXT NOT NULL, + ts_code TEXT NOT NULL, + name TEXT, + close REAL, + pct_change REAL, + amount REAL, + reason TEXT, + buy_seats INTEGER, + sell_seats INTEGER, + net_amount REAL, + PRIMARY KEY (trade_date, ts_code, reason) +); + +-- 7x24 快讯流（精确到秒） +CREATE TABLE IF NOT EXISTS news_flash ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content TEXT NOT NULL, + event_time TEXT NOT NULL, + source TEXT, + inserted_at TEXT NOT NULL +); + +-- 每日大事提醒（政策级） +CREATE TABLE IF NOT EXISTS macro_event ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_date TEXT NOT NULL, + event_time TEXT, + title TEXT NOT NULL, + event_type INTEGER, + source TEXT, + actual TEXT, + forecast TEXT, + prev TEXT, + inserted_at TEXT NOT NULL DEFAULT (datetime('now','localtime')) +); + +-- 全网财经新闻 +CREATE TABLE IF NOT EXISTS news_cn ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + content TEXT, + pub_date TEXT NOT NULL, + author TEXT, + source TEXT, + url TEXT, + keywords TEXT, + inserted_at TEXT NOT NULL +); + +-- 北向/主力资金流（日频） +CREATE TABLE IF NOT EXISTS money_flow ( + ts_code TEXT NOT NULL, + trade_date TEXT NOT NULL, + main_net_in REAL, + in_flow REAL, + out_flow REAL, + amount REAL, + pct_change REAL, + PRIMARY KEY (ts_code, trade_date) +); + +-- 全球指数（日频，外盘收盘后入库） +CREATE TABLE IF NOT EXISTS global_index ( + index_code TEXT NOT NULL, + trade_date TEXT NOT NULL, + close REAL, + pct_change REAL, + PRIMARY KEY (index_code, trade_date) +); + +-- 采集记录（断点续采用） +CREATE TABLE IF NOT EXISTS fetch_log ( + source TEXT NOT NULL, + param TEXT, + last_ts TEXT, + last_date TEXT, + updated_at TEXT NOT NULL, + PRIMARY KEY (source, param) +); +""" + +def init_db(): + with get_conn() as conn: + conn.executescript(SCHEMA_SQL) + print(f"[OK] Database initialized: {DB_PATH}") + + +# ───────────────────────────────────────────────────────────────────────────── +# 通用 upsert +# ───────────────────────────────────────────────────────────────────────────── + +def upsert_df( + df: pd.DataFrame, + table: str, + pk_cols: list[str] = None, # 暂未使用，to_sql(replace) 已处理主键冲突 + csv备份: bool = True +) -> int: + """主键冲突时 replace，等效于 upsert""" + if df is None or df.empty: + return 0 + # 自动追加 inserted_at / updated_at + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + if "inserted_at" in df.columns: + df["inserted_at"] = now + elif table in ("news_flash", "macro_event", "news_cn"): + df["inserted_at"] = now + + with get_conn() as conn: + # sqlite 不支持 df 直接 execute，需要逐行或用 executemany + df.to_sql(table, conn, if_exists="replace", index=False) + row_count = len(df) + + if csv备份: + to_csv(df, table) + print(f" [OK] {table}: {row_count} rows upserted") + return row_count + + +def upsert_log(source: str, param: str, last_ts: str = None, last_date: str = None): + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with get_conn() as conn: + conn.execute(""" + INSERT INTO fetch_log (source, param, last_ts, last_date, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(source, param) DO UPDATE SET + last_ts = COALESCE(excluded.last_ts, last_ts), + last_date = COALESCE(excluded.last_date, last_date), + updated_at = excluded.updated_at + """, (source, param, last_ts, last_date, now)) + + +def get_last_fetch(source: str, param: str = "") -> dict: + with get_conn() as conn: + row = conn.execute( + "SELECT last_ts, last_date, updated_at FROM fetch_log WHERE source=? AND param=?", + (source, param) + ).fetchone() + if row: + return {"last_ts": row[0], "last_date": row[1], "updated_at": row[2]} + return {} + + +if __name__ == "__main__": + init_db() + +# --------------------------------------------------------------------------- +# 保历史的 upsert（INSERT OR REPLACE，不清空表） +# --------------------------------------------------------------------------- + +_IDENT_RE = __import__('re').compile(r'^[A-Za-z0-9_]+$') + + +def _safe_ident(name: str) -> str: + """SQL 标识符白名单校验（列名/表名不可参数化，只允许字母数字下划线）""" + if not _IDENT_RE.match(str(name)): + raise ValueError("illegal identifier: %r" % (name,)) + return str(name) + + +def ensure_index(table: str, cols: list) -> None: + """按冲突列建唯一索引（幂等）""" + tab = _safe_ident(table) + col_ids = [_safe_ident(c) for c in cols] + idx = "ux_{}_{}".format(tab, "_".join(col_ids)) + with get_conn() as conn: + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {}({})".format( + idx, tab, ",".join(col_ids))) + + +def upsert_rows(df, table: str, conflict_cols: list) -> int: + """INSERT OR REPLACE 按 conflict_cols 去重，保留既有历史。值全部走 ? 占位符。""" + if df is None or df.empty: + return 0 + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + if "inserted_at" in df.columns: + df["inserted_at"] = now + tab = _safe_ident(table) + col_ids = [_safe_ident(c) for c in df.columns] + sql = "INSERT OR REPLACE INTO {}({}) VALUES ({})".format( + tab, ",".join(col_ids), ",".join("?" * len(col_ids))) + rows = [tuple(None if (isinstance(v, float) and v != v) else v + for v in tup) for tup in df.itertuples(index=False, name=None)] + with get_conn() as conn: + conn.executemany(sql, rows) + print(" [OK] {}: {} rows upserted (conflict on {})".format(tab, len(rows), conflict_cols)) + return len(rows) diff --git a/start_realtime.bat b/start_realtime.bat new file mode 100644 index 0000000..b963d10 --- /dev/null +++ b/start_realtime.bat @@ -0,0 +1,5 @@ +@echo off +title JQuant Realtime Analyzer +cd /d %~dp0 +python src\realtime\realtime_analyzer.py +pause diff --git a/test_fetch.py b/test_fetch.py new file mode 100644 index 0000000..333860a --- /dev/null +++ b/test_fetch.py @@ -0,0 +1,74 @@ +"""快速采集测试脚本""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent)) + +from src.fetcher import stock_api, lhb_fetcher, news_flash_fetcher, macro_event_fetcher +from src.storage.db import upsert_df, upsert_log, get_conn +from datetime import datetime, timedelta + +def test_daily(): + print("=== [1/4] 日K线（主要指数 90天） ===") + df = stock_api.batch_fetch_indices(days=90) + if df is not None and not df.empty: + upsert_df(df, "stock_daily") + print(f" 索引列表: {sorted(df['ts_code'].unique())}") + print(f" 总行数: {len(df)}") + # 验证最近一条 + latest = df.sort_values("trade_date").tail(3) + print(f" 最新3条:\n{latest[['ts_code','trade_date','close']].to_string(index=False)}") + else: + print(" [FAIL] 返回空") + +def test_lhb(): + print("\n=== [2/4] 龙虎榜（近10日，含上周五） ===") + # 上周五（最近确定有数据的交易日） + start = (datetime.today() - timedelta(days=10)).strftime("%Y-%m-%d") + end = datetime.today().strftime("%Y-%m-%d") + df = lhb_fetcher.fetch_lhb_range(start, end) + if df is not None and not df.empty: + upsert_df(df, "lhb_daily") + upsert_log("lhb_daily", "", last_date=end) + print(f" 总行数: {len(df)}") + print(f" 涵盖交易日: {sorted(df['trade_date'].unique())}") + print(f" 样例:\n{df[['trade_date','ts_code','name','reason']].head(5).to_string(index=False)}") + else: + print(" [WARN] 该时间段无龙虎榜数据（可能是非交易日）") + +def test_flash(): + print("\n=== [3/4] 7x24 快讯流（AkShare 财联社） ===") + df = news_flash_fetcher.fetch_akshare_news() + if df is not None and not df.empty: + upsert_df(df, "news_flash", csv备份=False) + print(f" 总行数: {len(df)}") + print(f" 样例:\n{df.head(3).to_string(index=False)}") + else: + print(" [WARN] 无快讯数据") + +def test_events(): + print("\n=== [4/4] 大事提醒 ===") + df = macro_event_fetcher.fetch_macro_calendar() + if df is not None and not df.empty: + upsert_df(df, "macro_event") + print(f" 总行数: {len(df)}") + print(f" 样例:\n{df.head(5).to_string(index=False)}") + else: + print(" [WARN] 无数据（可能接口报表名需调整）") + +def db_summary(): + print("\n=== 数据库现状 ===") + with get_conn() as conn: + tables = ["stock_daily","lhb_daily","news_flash","macro_event","news_cn"] + for t in tables: + try: + cnt = conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] + print(f" {t}: {cnt} rows") + except Exception: + print(f" {t}: 0 rows (table may not exist)") + +if __name__ == "__main__": + test_daily() + test_lhb() + test_flash() + test_events() + db_summary() diff --git a/test_lhb.py b/test_lhb.py new file mode 100644 index 0000000..0fffc25 --- /dev/null +++ b/test_lhb.py @@ -0,0 +1,45 @@ +import sys; sys.path.insert(0, 'C:/Users/lookt/a_stock_timeline') +from src.fetcher import lhb_fetcher +import requests + +# Test LHB for last Friday +for date in ['2026-09-12', '2026-09-11']: + print(f"\n=== LHB {date} ===") + df = lhb_fetcher.fetch_lhb_date(date) + print(f" rows: {len(df)}") + if not df.empty: + print(df[['trade_date','ts_code','name','reason']].head(3).to_string(index=False)) + +# If empty, check raw API response +print("\n=== Raw API check ===") +url = 'https://datacenter-web.eastmoney.com/api/data/v1/get' +params = { + 'reportName': 'RPT_LHB_ALL_STOCKS', + 'columns': 'ALL', + 'filter': "(TRADE_DATE='2026-09-12')", + 'pageNumber': 1, + 'pageSize': 3, + 'sortTypes': '-1', + 'sortColumns': 'SINGLE_NET_AMT', + 'source': 'WEB', + 'client': 'WEB', +} +headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', + 'Referer': 'https://data.eastmoney.com/', + 'Accept': 'application/json', +} +r = requests.get(url, params=params, headers=headers, timeout=15) +print('status:', r.status_code) +resp = r.json() +print('success:', resp.get('success')) +print('message:', resp.get('message')) +result = resp.get('result') +if result: + print('pages:', result.get('pages')) + data = result.get('data') or [] + print('data count:', len(data)) + if data: + print('first row keys:', list(data[0].keys())[:8]) +else: + print('result is None or empty') diff --git a/test_lhb2.py b/test_lhb2.py new file mode 100644 index 0000000..ac0ba0f --- /dev/null +++ b/test_lhb2.py @@ -0,0 +1,41 @@ +import requests + +EM_BASE = "https://datacenter-web.eastmoney.com/api/data/v1/get" + +headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Referer": "https://data.eastmoney.com/", + "Accept": "application/json", +} + +# 已知可能的龙虎榜报表名变体 +report_names = [ + "RPT_LHB_ALL_STOCKS", + "RPT_LHB_ALL", # 龙虎榜概览 + "RPT_LHB_STOCK", # 个股龙虎榜 + "RPT_LHB_BILLBOARD", # 可能的英文名 + "RPT_STOCK_LHB", # 倒装 + "LHB_ALL_STOCKS", # 无前缀 +] + +for report in report_names: + params = { + "reportName": report, + "columns": "ALL", + "filter": "(TRADE_DATE='2026-09-12')", + "pageNumber": 1, + "pageSize": 2, + "source": "WEB", + "client": "WEB", + } + try: + r = requests.get(EM_BASE, params=params, headers=headers, timeout=10) + resp = r.json() + if resp.get("success"): + result = resp.get("result") or {} + data = result.get("data") or [] + print(f"[OK] {report}: {len(data)} rows, keys={list(data[0].keys())[:5] if data else 'empty'}") + else: + print(f"[FAIL] {report}: {resp.get('message', 'unknown error')}") + except Exception as e: + print(f"[ERR] {report}: {e}")