# stock-data-hub

**Repository Path**: tin555/stock-data-hub

## Basic Information

- **Project Name**: stock-data-hub
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-02-25
- **Last Updated**: 2026-02-25

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# DuckD - DuckDB 数据同步工具

一个基于 DuckDB 和 APScheduler 的数据同步工具，支持自定义策略和执行顺序控制。

## 功能特点

- 支持自定义数据同步策略
- 支持定时执行和立即同步
- 支持通过优先级控制同步顺序（PRIORITY 值越小优先级越高）
- 支持同步前清空表数据（SYNCDELETE=True）
- 支持启用/禁用策略（ENABLE=True/False）
- 基于 DuckDB 进行数据存储

## 项目结构

```
duckd/
├── config.py           # 配置文件
├── core/               # 核心模块
│   ├── duckdb_client.py    # DuckDB 客户端
│   ├── scheduler.py        # 调度器
│   └── strategy_loader.py  # 策略加载器
├── data/               # 数据存储
│   └── stock_data.db       # DuckDB 数据库文件
├── logs/               # 日志文件
├── main.py             # 程序入口
└── strategies/         # 策略文件
    └── stock_daily.py      # 股票日线数据策略示例
```

## 使用说明

### 1. 运行程序

```bash
python main.py
```

### 2. 可用命令

- `sync` - 立即同步所有策略数据（按优先级排序执行）
- `exit` - 退出程序
- `help` - 查看命令帮助

### 3. 创建自定义策略

在 `strategies` 目录下创建新的 Python 文件，按照以下模板定义策略：

```python
import pandas as pd

# 策略配置（使用小写键名的字典统一管理参数）
strategy_config = {
    "strategy_name": "your_strategy_name",  # 唯一标识
    "table_name": "your_table_name",        # 目标表名
    "table_struct": [                      # 表结构（数组包字典格式）
        {"name": "field1", "type": "VARCHAR", "comment": "字段1备注", "primary_key": True},
        {"name": "field2", "type": "FLOAT", "comment": "字段2备注", "primary_key": False},
        # ...更多字段
    ],
    "execute_mode": "daily",                 # 执行模式（daily/date_range）
    "cron_expr": "0 0 3 * * ?",               # 执行时间（Quartz风格cron表达式，格式：秒 分 时 日 月 周）
    "priority": 100,                         # 优先级（值越小越先执行，可选，默认为999）
    "sync_delete": False,                     # 同步前是否清空表数据（可选，默认为False）
    "enable": True,                          # 是否启用策略（可选，默认为True）
    # 可选配置（按需取消注释）
    # "start_date": "2023-01-01",              # 开始日期（date_range模式下使用）
    # "end_date": "2023-12-31",                # 结束日期（date_range模式下使用）
}

def download_data(params: dict) -> pd.DataFrame:
    """下载数据的核心逻辑（必须实现）"""
    # params包含系统传入的参数
    execute_date = params["execute_date"]
    
    # 实现数据下载逻辑
    # ...
    
    # 返回DataFrame格式的数据
    return pd.DataFrame()
```

## 优先级、同步删除和启用控制说明

- **优先级控制**：通过在策略文件中定义 `PRIORITY` 变量来控制策略的执行顺序。`PRIORITY` 值越小，策略的优先级越高，会先执行。如果不定义 `PRIORITY` 变量，默认优先级为 999。

- **同步前清空表**：通过在策略文件中定义 `SYNCDELETE` 变量来控制是否在同步前清空表数据。设置为 `True` 时，每次同步前会清空该策略对应的表；设置为 `False` 时，不会清空表，新数据会追加或忽略（根据主键冲突规则）。如果不定义 `SYNCDELETE` 变量，默认值为 `False`。


## 依赖安装

```bash
pip install pandas duckdb apscheduler
```

## 快速自检：AkShare 可用性

提供了一个健康检查脚本，快速验证常用接口是否可访问。

```bash
python scripts/healthcheck.py
```

预期输出示例：

```
[OK] tool_trade_date_hist_sina rows=XXXX
[OK] stock_info_a_code_name rows=XXXX
[OK] stock_zh_index_daily_tx rows=XXXX
AkShare healthcheck passed.
```

## 运行测试（pytest）

项目内含网络依赖的简单测试样例，默认会请求 AkShare：

```bash
pip install -U pytest
pytest -q
```

如在 CI 或无网络环境，可跳过网络测试：

```bash
SKIP_NETWORK_TESTS=1 pytest -q
```

## 可用策略清单（节选）

- 腾讯-指数/股票历史行情：`strategies/腾讯指数历史数据_tx.py`
  - 接口：`ak.stock_zh_index_daily_tx(symbol="sh000919")`
  - 特性：通过 DuckDB 表 `stock_zh_index_spot_em` 自动加载指数代码，并映射为腾讯格式 `sh/sz` 前缀；支持 `params["symbols"]` 覆盖。
- 指数快照（东方财富）：`strategies/指数数据列表.py`
  - 表：`stock_zh_index_spot_em`
- 涨停/跌停股票池：`strategies/涨停股池.py`、`strategies/跌停股票池.py`
- 指数目录/成份/历史：`strategies/指数.py`、`strategies/指数成分股.py`、`strategies/指数历史数据.py`
- 交易日日历：`strategies/股票交易日数据.py`

## 工具与实践

- `core/strategy_utils.py`
  - `ak_call(func, retries=3, backoff_sec=1.5, **kwargs)`：AkShare 调用重试与退避。
  - `insert_with_checks(db_client, table, df, pk=[...], min_rows=...)`：基础数据质量检查后落库。
  - `iter_trade_days(conn, start_date, end_date, inclusive_start=False, inclusive_end=True, as_format='timestamp')`：标准化交易日区间迭代。