# vnpy_wjf

**Repository Path**: ldsrc/vnpy_wjf

## Basic Information

- **Project Name**: vnpy_wjf
- **Description**: vnpy 接入网际风行情
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2026-05-06
- **Last Updated**: 2026-05-23

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# vnpy_wjf — 网际风行情 Datafeed

VeighNa 框架的网际风行情数据服务，通过 TCP/JSON 协议对接本地 `wjf_transfer.exe`（C++）转发服务。

## 数据链路

```
wjf_transfer.exe (C++) ←→ WjfClient (TCP/JSON) ←→ WjfDatafeed (BaseDatafeed) ←→ vnpy 框架
```

## 项目结构

```
vnpy_wjf/
├── pyproject.toml
├── README.md
├── vnpy_wjf/
│   ├── __init__.py              # 导出 WjfDatafeed as Datafeed
│   ├── wjf_datafeed.py          # Datafeed 主实现
│   ├── wjf_client.py            # TCP/JSON 客户端（连接 wjf_transfer）
│   └── api/
│       └── __init__.py           # 预留
├── tests/
│   ├── test_quick.py            # 最简连通性测试
│   ├── test_client.py           # 独立测试 WjfClient
│   └── test_datafeed.py         # 独立测试 WjfDatafeed
└── examples/
    └── wjf_demo/
        └── run.py               # vnpy 框架集成示例
```

## 安装

```bash
pip install -e /path/to/vnpy_wjf
```

## 调试方法

### 第一步：连通性测试（不依赖 vnpy）

```bash
python tests/test_quick.py
```

如果此脚本失败，说明 `wjf_transfer.exe` 未启动或端口配置错误。

### 第二步：测试 TCP 客户端

```bash
python tests/test_client.py
```

这个脚本独立测试 `WjfClient`，绕开 vnpy 框架，直接验证：
- TCP 连接是否正常
- K线/Tick/股票列表查询是否返回正确数据
- 实时行情订阅回调是否触发
- JSON 编解码是否正常

### 第三步：测试 Datafeed

```bash
python tests/test_datafeed.py
```

测试 `WjfDatafeed`，验证：
- `init()` 连接是否成功
- `query_bar_history()` 返回的 BarData 是否正确
- 各种 K 线周期（日/分钟/小时）是否支持

### 第四步：在 vnpy 框架内运行

```bash
python examples/wjf_demo/run.py
```

### 调试技巧

**添加临时打印**：在 `wjf_datafeed.py` 的 `query_bar_history()` 中打印原始返回数据：
```python
raw = self.client.query_kline(...)
print(f"[DEBUG] raw type={type(raw)} len={len(raw) if raw else 'None'}")  # 临时
if raw:
    print(f"[DEBUG] first item keys={list(raw[0].keys())}")  # 检查字段名
```

**模拟 wjf_transfer 响应**：如果 wjf_transfer 不可用，可以在测试脚本中 mock 客户端：
```python
from unittest.mock import MagicMock
client = MagicMock()
client.query_kline.return_value = [{"datetime": "2025-05-16", "open": 10, ...}]
```

**检查 TCP 原始数据**：在 `wjf_client.py` 的 `_recv_loop()` 中打印接收到的原始字节：
```python
print(f"[RAW] {chunk[:200]}")  # 仅前 200 字节
```

**常见问题**：
| 现象 | 可能原因 | 检查方法 |
|------|---------|---------|
| 连接被拒绝 | wjf_transfer 未启动 | 运行 `test_quick.py` |
| 查询返回空 | 合约代码格式不对 | 确认 label 是 `SH600000` 而非 `600000.SSE` |
| JSON 解码失败 | 换行符不一致 | 检查 `\r\n` vs `\n` |
| BarData 日期错误 | 时间格式匹配失败 | 检查 `_parse_datetime()` 的格式列表 |
| 数据量过大超时 | count 参数过大 | 调小 `count` 或增大 `timeout` |

## 在 vnpy 框架中使用

在 `examples/veighna_trader/run.py` 或自定义启动脚本中：

```python
from vnpy.trader.setting import SETTINGS

# 配置 datafeed
SETTINGS["datafeed.name"] = "wjf"
SETTINGS["datafeed.host"] = "127.0.0.1"
SETTINGS["datafeed.port"] = 16888

# vnpy 框架自动通过 get_datafeed() 加载 vnpy_wjf.WjfDatafeed
```