Compare commits

...

10 Commits

Author SHA1 Message Date
王鹏
b1eea9f5b4 feat: 优化界面并补充网格成交分析 2026-07-14 16:21:32 +08:00
王鹏
8b51463387 refactor: use tabs for selected instrument tables 2026-07-09 11:57:15 +08:00
王鹏
5d66423895 feat: show open grid lots 2026-07-09 11:17:51 +08:00
王鹏
fe11929aa0 docs: design open grid lots 2026-07-09 11:14:30 +08:00
王鹏
6d30d48c33 feat: add grid level suggestions 2026-07-09 10:50:25 +08:00
王鹏
15636fb910 fix: separate trade price from realtime price 2026-07-09 10:45:27 +08:00
王鹏
5bbc4e607a docs: plan grid level suggestions 2026-07-09 10:44:47 +08:00
王鹏
0c278d07be docs: design grid level suggestions 2026-07-09 10:17:35 +08:00
王鹏
659846fe18 fix: refresh quotes without blocking UI 2026-07-09 10:01:15 +08:00
王鹏
1e8858ac04 fix: retry transient Tencent quote failures 2026-07-08 22:09:17 +08:00
24 changed files with 2932 additions and 133 deletions

203
README.md
View File

@@ -1,59 +1,216 @@
# Grid Trading
# Grid Trading:本地网格交易管理器
本项目是一个本地运行的 Python GUI 网格交易管理工具,第一阶段聚焦 A股/ETF 的账户、标的手工成交持仓成本、网格收益回本价计算
一个使用 Python 和 PySide6 开发的本地桌面工具,用于管理 A 股、ETF 的账户、标的手工成交,并辅助计算持仓成本、网格收益回本价和网格档位
## 第一阶段功能
> 本项目只提供交易记录与策略辅助功能,不会连接券商账户,也不会自动下单。所有计算结果仅供参考,不构成投资建议。
- 创建/编辑本地账户和初始资金。
- 添加 A股/ETF 标的,维护交易单位和手动估值价格。
- 配置默认网格策略模板。
- 手动录入、编辑、删除买入/卖出成交。
- 自动计算持仓数量、T+1 可用数量、持仓成本、已实现盈亏、累计网格利润、持仓回本价、账户回本价。
- 点击“刷新行情”从腾讯接口获取实时价格,并用实时价更新持仓市值和账户摘要。
- 使用 SQLite 本地保存数据,默认路径为 `data/grid_trading.db`
## 功能特性
## 行情说明
- **账户概览**:展示账户权益、现金、持仓市值、浮动盈亏和资金使用率。
- **标的管理**:维护股票或 ETF 的代码、名称、市场、交易单位及零股规则。
- **成交管理**:手工录入、编辑和删除买卖记录,并区分底仓、网格仓和其他仓位。
- **费用估算**:按内置费率估算佣金、印花税和过户费,估算后仍可手工调整。
- **持仓分析**计算总持仓、T+1 可用数量、剩余成本、已实现盈亏、浮动盈亏及网格利润。
- **回本价计算**:同时提供持仓回本价与账户回本价,便于观察网格收益对成本的影响。
- **实时行情**:异步获取腾讯行情,刷新过程中不会阻塞主界面。
- **网格建议**:根据实时价格和默认策略,生成最多 100 档的买入价、建议数量、卖出价与预计毛利润。
- **待卖网格**:按批次展示尚未卖出的网格买入及其目标卖出价、当前状态。
- **成交配对**:使用 FIFO先进先出将网格卖出与历史网格买入配对一笔卖出可拆分匹配多笔买入。
- **本地存储**:账户、标的、策略和成交数据均保存在 SQLite 数据库中。
行情刷新使用腾讯接口:
## 技术栈
```text
http://qt.gtimg.cn/q=<symbol>
- Python 3.11+
- PySide6
- SQLite
- pytest
## 快速开始
### 1. 进入项目目录
```powershell
cd Grid_Trading
```
例如 `000001` 会自动转换为 `sz000001``600000` 会自动转换为 `sh600000`
请先下载或克隆项目,再进入项目根目录
实时行情只保存在当前程序内存中,用于显示现价、持仓市值、浮动盈亏、总资产和资金使用率;不会覆盖标的里的手动价格,也不会写入 SQLite。关闭软件后再次打开需要重新点击“刷新行情”。
### 2. 创建虚拟环境并安装依赖
## 开发环境
Windows PowerShell
```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install -e ".[dev]"
pytest
```
## 启动
macOS / Linux
```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -e ".[dev]"
```
### 3. 启动程序
```powershell
python -m grid_trading.app
```
也可以指定数据库路径
安装后也可以使用命令行入口
```powershell
grid-trading
```
默认数据库位置为 `data/grid_trading.db`。如需使用其他数据库文件:
```powershell
python -m grid_trading.app --db data/dev.db
```
## 验证
首次启动时,程序会自动创建:
- 初始资金为 `0` 的“默认账户”;
- 间距为 `3%`、每格金额为 `5000` 元的“默认网格模板”;
- 数据库及所需数据表。
## 基本使用流程
1. 点击“账户设置”,填写账户名称和初始资金。
2. 点击“添加标的”,填写证券代码、名称、市场和交易单位。
3. 点击“策略设置”,调整网格间距、每格金额等参数。
4. 点击“录入成交”,选择买卖方向、仓位分组并填写实际成交信息。
5. 点击“刷新行情”,获取现价并更新持仓市值、浮动盈亏和账户摘要。
6. 在持仓表中选择一个标的,查看“网格档位”“待卖网格”“配对明细”和“最近成交”。
## 行情说明
行情数据来自腾讯接口:
```text
http://qt.gtimg.cn/q=<symbol>
```
程序会根据六位证券代码推断交易所前缀,例如:
| 输入代码 | 请求代码 |
| --- | --- |
| `000001` | `sz000001` |
| `600000` | `sh600000` |
| `510300` | `sh510300` |
| `430047` | `bj430047` |
需要注意:
- 行情刷新需要网络连接,单次请求超时时间为 5 秒,失败后会自动重试一次。
- 行情快照只保存在当前进程内存中,不会写入 SQLite。
- 重新启动程序后,需要再次点击“刷新行情”。
- 未刷新行情时,不使用手工价格或最近成交价代替现价,因此市值、浮动盈亏和网格档位可能为空。
## 计算口径
### 持仓与盈亏
- 买入成本 = 成交金额 + 交易费用。
- 卖出净收入 = 成交金额 - 交易费用。
- 同一仓位分组内的卖出成本按移动平均成本释放。
- 已实现盈亏 = 卖出净收入 - 被释放的持仓成本。
- 网格利润只统计“网格”分组的已实现盈亏。
- T+1 可用数量会扣除计算当日的买入数量;卖出数量不得超过可用数量,也不得超过对应分组的持仓数量。
- 修改或删除历史成交后,程序会重新校验后续成交,避免形成负持仓。
### 回本价与账户摘要
- 持仓回本价 =(剩余持仓成本 - 已实现网格利润)/ 当前持仓数量。
- 账户回本价 = 标的累计净投入 / 当前持仓数量。
- 账户权益 = 现金 + 持仓市值。
- 资金使用率 = 持仓市值 /(持仓市值 + 非负现金);现金为负时最高显示为 100%。
### 网格档位
第 1 档买入价按下式计算,之后逐档复合递减:
```text
买入价 = 上一档参考价 × (1 - 网格间距)
卖出价 = 买入价 × (1 + 网格间距)
建议数量 = 每格金额 ÷ 买入价,并按交易单位向下取整
```
档位表中的“预计单轮毛利润”不扣除交易费用,且建议结果只用于展示,不会自动生成成交记录或下单。
### 待卖网格与配对明细
- 只有“网格”分组的成交参与待卖批次和配对计算。
- 网格卖出按照成交日期和记录顺序,优先抵消最早的网格买入。
- 待卖批次状态包括“未刷新行情”“未到价”和“可卖”。
- 配对明细展示历史买卖的对应关系和毛利润,不扣除费用。
## 默认费用参数
点击成交窗口中的“估算费用”时,程序使用以下默认参数:
| 项目 | 默认值 | 说明 |
| --- | ---: | --- |
| 佣金 | 0.025% | 最低 5 元,买卖双向收取 |
| 印花税 | 0.05% | 仅卖出时收取 |
| 过户费 | 0.001% | 买卖双向收取 |
不同券商、市场和交易品种的实际费用可能不同,请以实际交割单为准,并在保存成交前手工修正。
## 数据与隐私
- 默认数据文件:`data/grid_trading.db`
- 账户、成交和策略数据仅存放在本机;刷新行情时只会向腾讯行情接口发送证券代码。
- `data/*.db``data/*.sqlite``data/backups/` 已在 `.gitignore` 中忽略。
- 建议定期备份数据库文件;复制数据库前请先退出程序,避免备份到尚未提交的写入状态。
## 项目结构
```text
Grid_Trading/
├── src/grid_trading/
│ ├── app.py # 命令行入口
│ ├── config.py # 默认路径与费用配置
│ ├── domain/ # 数据模型、持仓和网格计算
│ ├── market/ # 腾讯行情适配
│ ├── services/ # 业务编排与校验
│ ├── storage/ # SQLite 数据库与仓储层
│ └── ui/ # PySide6 主窗口、对话框与格式化
├── tests/ # 自动化测试
├── docs/ # 设计说明与开发计划
├── data/ # 本地数据库目录
└── pyproject.toml # 项目及依赖配置
```
## 测试
运行完整测试:
```powershell
pytest -v
python -m grid_trading.app --help
```
如果只想确认 GUI 能构造起来,不打开真实窗口,可以运行测试里的 offscreen smoke
只验证界面可在无显示环境下正常构造
```powershell
pytest tests/test_ui.py -v
```
检查命令行参数:
```powershell
python -m grid_trading.app --help
```
## 当前限制
- 当前定位为单机、单用户的手工交易管理工具。
- 不包含券商登录、自动下单、条件单或实盘同步功能。
- 行情接口不是正式授权的数据服务,其可用性、实时性和准确性不作保证。
- 网格建议未考虑手续费、滑点、涨跌停、停牌和流动性等实际交易约束。
- 使用本项目产生的任何交易决策与风险均由使用者自行承担。

View File

@@ -15,11 +15,11 @@
- Create `src/grid_trading/market/__init__.py`: market package marker.
- Create `src/grid_trading/market/tencent.py`: Tencent symbol inference, response parsing, and HTTP quote provider.
- Modify `src/grid_trading/domain/models.py`: add `QuoteSnapshot`.
- Modify `src/grid_trading/domain/calculations.py`: let quote snapshots override manual price when computing positions.
- Modify `src/grid_trading/domain/calculations.py`: let quote snapshots supply current price when computing positions.
- Modify `src/grid_trading/services/trading_service.py`: add quote provider injection, `refresh_quotes`, and quote cache use in summaries.
- Modify `src/grid_trading/ui/main_window.py`: make the toolbar refresh action fetch Tencent quotes and display quote refresh errors.
- Create `tests/test_tencent_quotes.py`: parser and symbol inference tests.
- Modify `tests/test_calculations.py`: verify quote price overrides manual price.
- Modify `tests/test_calculations.py`: verify quote price supplies current price.
- Modify `tests/test_services.py`: verify service refresh uses fake quote provider and summaries use realtime price.
## Tasks
@@ -98,7 +98,7 @@ git commit -m "feat: add Tencent quote parser"
- [ ] **Step 1: Write failing calculation and service tests**
Add a calculation test where an `Instrument` has `manual_price=Decimal("9")`, a `QuoteSnapshot(price=Decimal("10"))` is passed, and `PositionSummary.current_price` becomes `10` with `price_source == "tencent"`.
Add a calculation test where a `QuoteSnapshot(price=Decimal("10"))` is passed, and `PositionSummary.current_price` becomes `10` with `price_source == "tencent"`.
Add a service test with a fake provider:

View File

@@ -0,0 +1,419 @@
# Grid Level Suggestions Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add a GUI grid-level table that generates buy/sell grid suggestions from Tencent current price and the existing strategy percentage spacing.
**Architecture:** Put the grid formula in a pure domain module, expose it through `TradingService`, and render it in the selected-instrument area of the PySide6 main window. The feature uses existing strategy template fields (`grid_spacing_pct`, `amount_per_grid`) and existing quote-backed `PositionSummary.current_price`.
**Tech Stack:** Python 3.11, Decimal, PySide6, SQLite repository/service pattern, pytest.
---
## File Structure
- Create `src/grid_trading/domain/grid_levels.py`: pure grid-level calculation, independent of PySide6 and SQLite.
- Modify `src/grid_trading/domain/models.py`: add `GridLevelSuggestion`.
- Modify `src/grid_trading/services/trading_service.py`: add `get_grid_level_suggestions(instrument_id, levels=10)`.
- Modify `src/grid_trading/ui/main_window.py`: add a “网格档位” table and a non-persisted level-count spin box.
- Create `tests/test_grid_levels.py`: domain formula tests.
- Modify `tests/test_services.py`: service integration tests for quote-backed suggestions and missing-quote empty state.
- Modify `tests/test_ui.py`: GUI smoke test for the grid-level table.
- Modify `README.md`: document the grid-level table behavior.
## Tasks
### Task 1: Domain Grid-Level Formula
**Files:**
- Create: `src/grid_trading/domain/grid_levels.py`
- Modify: `src/grid_trading/domain/models.py`
- Test: `tests/test_grid_levels.py`
- [ ] **Step 1: Write failing formula tests**
Add tests that exercise the wished-for API:
```python
from decimal import Decimal
import pytest
from grid_trading.domain.grid_levels import generate_grid_levels
def test_generate_grid_levels_compounds_percentage_spacing_and_rounds_values():
levels = generate_grid_levels(
current_price=Decimal("10.00"),
spacing=Decimal("0.03"),
amount_per_grid=Decimal("10000"),
lot_size=100,
levels=3,
)
assert [item.level for item in levels] == [1, 2, 3]
assert [item.buy_price for item in levels] == [Decimal("9.70"), Decimal("9.41"), Decimal("9.13")]
assert [item.buy_amount for item in levels] == [Decimal("10000.00")] * 3
assert [item.suggested_quantity for item in levels] == [1000, 1000, 1000]
assert [item.actual_investment for item in levels] == [
Decimal("9700.00"),
Decimal("9410.00"),
Decimal("9130.00"),
]
assert [item.sell_price for item in levels] == [Decimal("9.99"), Decimal("9.69"), Decimal("9.40")]
assert [item.estimated_gross_profit for item in levels] == [
Decimal("290.00"),
Decimal("280.00"),
Decimal("270.00"),
]
def test_generate_grid_levels_uses_zero_quantity_when_amount_cannot_buy_one_lot():
[level] = generate_grid_levels(
current_price=Decimal("10.00"),
spacing=Decimal("0.03"),
amount_per_grid=Decimal("500"),
lot_size=100,
levels=1,
)
assert level.buy_price == Decimal("9.70")
assert level.suggested_quantity == 0
assert level.actual_investment == Decimal("0.00")
assert level.estimated_gross_profit == Decimal("0.00")
@pytest.mark.parametrize(
("current_price", "spacing", "amount_per_grid", "lot_size", "levels"),
[
(Decimal("0"), Decimal("0.03"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("0"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("1"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("0.03"), Decimal("0"), 100, 10),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 0, 10),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 100, 0),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 100, 101),
],
)
def test_generate_grid_levels_validates_inputs(current_price, spacing, amount_per_grid, lot_size, levels):
with pytest.raises(ValueError):
generate_grid_levels(
current_price=current_price,
spacing=spacing,
amount_per_grid=amount_per_grid,
lot_size=lot_size,
levels=levels,
)
```
- [ ] **Step 2: Run formula tests and verify red**
Run:
```powershell
pytest tests/test_grid_levels.py -q
```
Expected: FAIL because `grid_trading.domain.grid_levels` does not exist.
- [ ] **Step 3: Implement model and pure calculation**
Add `GridLevelSuggestion` to `models.py`:
```python
@dataclass(frozen=True)
class GridLevelSuggestion:
level: int
buy_price: Decimal
buy_amount: Decimal
suggested_quantity: int
actual_investment: Decimal
sell_price: Decimal
estimated_gross_profit: Decimal
```
Create `grid_levels.py` with:
```python
from __future__ import annotations
from decimal import Decimal
from grid_trading.domain.calculations import money, price
from grid_trading.domain.models import GridLevelSuggestion
def generate_grid_levels(
*,
current_price: Decimal,
spacing: Decimal,
amount_per_grid: Decimal,
lot_size: int,
levels: int,
) -> list[GridLevelSuggestion]:
if current_price <= 0:
raise ValueError("现价必须大于 0")
if spacing <= 0 or spacing >= 1:
raise ValueError("网格间距必须大于 0 且小于 100%")
if amount_per_grid <= 0:
raise ValueError("每格金额必须大于 0")
if lot_size <= 0:
raise ValueError("交易单位必须大于 0")
if levels < 1 or levels > 100:
raise ValueError("档数必须在 1 到 100 之间")
suggestions: list[GridLevelSuggestion] = []
buy_price = price(current_price * (Decimal("1") - spacing))
for level in range(1, levels + 1):
quantity = int(amount_per_grid / buy_price) // lot_size * lot_size
actual_investment = money(buy_price * Decimal(quantity))
sell_price = price(buy_price * (Decimal("1") + spacing))
estimated_gross_profit = money((sell_price - buy_price) * Decimal(quantity))
suggestions.append(
GridLevelSuggestion(
level=level,
buy_price=buy_price,
buy_amount=money(amount_per_grid),
suggested_quantity=quantity,
actual_investment=actual_investment,
sell_price=sell_price,
estimated_gross_profit=estimated_gross_profit,
)
)
buy_price = price(buy_price * (Decimal("1") - spacing))
return suggestions
```
- [ ] **Step 4: Run formula tests and verify green**
Run:
```powershell
pytest tests/test_grid_levels.py -q
```
Expected: PASS.
### Task 2: Service API
**Files:**
- Modify: `src/grid_trading/services/trading_service.py`
- Test: `tests/test_services.py`
- [ ] **Step 1: Write failing service tests**
Add tests:
```python
def test_service_generates_grid_level_suggestions_from_realtime_price(tmp_path):
service = TradingService(tmp_path / "grid.db", quote_provider=FakeQuoteProvider())
service.ensure_defaults()
account = service.get_active_account()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("4.00"),
quantity=1000,
trade_group=TradeGroup.BASE,
)
)
service.refresh_quotes()
levels = service.get_grid_level_suggestions(instrument.id, levels=2)
assert [item.buy_price for item in levels] == [Decimal("4.00"), Decimal("3.88")]
assert [item.sell_price for item in levels] == [Decimal("4.12"), Decimal("4.00")]
assert [item.buy_amount for item in levels] == [Decimal("5000.00"), Decimal("5000.00")]
assert [item.suggested_quantity for item in levels] == [1200, 1200]
def test_service_returns_empty_grid_levels_without_realtime_price(tmp_path):
service = TradingService(tmp_path / "grid.db")
service.ensure_defaults()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
assert service.get_grid_level_suggestions(instrument.id, levels=10) == []
```
- [ ] **Step 2: Run service tests and verify red**
Run:
```powershell
pytest tests/test_services.py::test_service_generates_grid_level_suggestions_from_realtime_price tests/test_services.py::test_service_returns_empty_grid_levels_without_realtime_price -q
```
Expected: FAIL because `get_grid_level_suggestions` does not exist.
- [ ] **Step 3: Implement service API**
Import `generate_grid_levels` and `GridLevelSuggestion`, then add:
```python
def get_grid_level_suggestions(
self,
instrument_id: int,
*,
levels: int = 10,
) -> list[GridLevelSuggestion]:
instrument = self._require_instrument(instrument_id)
position = next(
(
item
for item in self.get_position_summaries()
if item.instrument_id == instrument_id
),
None,
)
if position is None or position.current_price is None:
return []
template = self.get_default_strategy_template()
return generate_grid_levels(
current_price=position.current_price,
spacing=template.grid_spacing_pct,
amount_per_grid=template.amount_per_grid,
lot_size=instrument.lot_size,
levels=levels,
)
```
- [ ] **Step 4: Run service tests and verify green**
Run:
```powershell
pytest tests/test_services.py -q
```
Expected: PASS.
### Task 3: GUI Grid-Level Table
**Files:**
- Modify: `src/grid_trading/ui/main_window.py`
- Test: `tests/test_ui.py`
- [ ] **Step 1: Write failing GUI smoke test**
Add:
```python
def test_main_window_contains_grid_level_table(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QGroupBox
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.grid_levels_table.columnCount() == 7
assert any(group.title() == "网格档位" for group in window.findChildren(QGroupBox))
window.close()
service.close()
app.processEvents()
```
- [ ] **Step 2: Run GUI smoke test and verify red**
Run:
```powershell
pytest tests/test_ui.py::test_main_window_contains_grid_level_table -q
```
Expected: FAIL because `grid_levels_table` does not exist.
- [ ] **Step 3: Implement GUI table**
Add `GRID_LEVEL_COLUMNS`, import `QSpinBox`, create `self.grid_levels_table`, `self.grid_levels_count_edit`, and `self.grid_levels_hint` in `_build_ui`. Add helper methods:
```python
def _refresh_grid_levels(self, position: PositionSummary | None) -> None:
if position is None:
self.grid_levels_hint.setText("-")
self._fill_grid_levels_table([])
return
if position.current_price is None:
self.grid_levels_hint.setText("请先刷新行情")
self._fill_grid_levels_table([])
return
levels = self.service.get_grid_level_suggestions(
position.instrument_id,
levels=self.grid_levels_count_edit.value(),
)
self.grid_levels_hint.setText("")
self._fill_grid_levels_table(levels)
def _fill_grid_levels_table(self, levels) -> None:
self.grid_levels_table.setRowCount(len(levels))
for row, level in enumerate(levels):
values = [
str(level.level),
format_price(level.buy_price),
format_money(level.buy_amount),
format_quantity(level.suggested_quantity),
format_money(level.actual_investment),
format_price(level.sell_price),
format_money(level.estimated_gross_profit),
]
for column, value in enumerate(values):
self.grid_levels_table.setItem(row, column, QTableWidgetItem(value))
```
Call `_refresh_grid_levels(position)` from `_refresh_details`.
- [ ] **Step 4: Run GUI tests and verify green**
Run:
```powershell
pytest tests/test_ui.py -q
```
Expected: PASS.
### Task 4: Docs And Verification
**Files:**
- Modify: `README.md`
- [ ] **Step 1: Update README**
Add a first-phase bullet for the grid-level table and note that it depends on Tencent current price.
- [ ] **Step 2: Run all tests**
Run:
```powershell
pytest -q
```
Expected: all tests pass.
- [ ] **Step 3: Run CLI smoke**
Run:
```powershell
python -m grid_trading.app --help
```
Expected: help text prints normally.
## Self-Review
- Spec coverage: formula, service API, GUI table, no quote empty state, non-persistent levels count, and docs are covered.
- Placeholder scan: no unresolved placeholder text is intentionally left.
- Type consistency: `GridLevelSuggestion`, `generate_grid_levels`, and `get_grid_level_suggestions` names match across tasks.

View File

@@ -0,0 +1,200 @@
# GUI Visual Refresh Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Refresh the PySide6 main window into a clean light desktop-tool style and apply A-share profit/loss colors to all PnL-like fields.
**Architecture:** Keep the existing `MainWindow` layout and data flow. Add focused UI helpers in `main_window.py` for theme setup, table setup, and PnL styling, then cover them with lightweight offscreen GUI tests.
**Tech Stack:** Python 3.11+, PySide6, pytest.
---
## File Structure
- Modify `src/grid_trading/ui/main_window.py`: apply QSS theme, assign object names, centralize table setup, and style PnL labels/items.
- Modify `tests/test_ui.py`: add offscreen GUI tests for theme markers and profit/loss item styling.
- No domain, service, storage, or database changes.
### Task 1: Add Failing GUI Tests
**Files:**
- Test: `tests/test_ui.py`
- [ ] **Step 1: Write failing tests for theme markers and PnL colors**
```python
def test_main_window_applies_light_theme(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.centralWidget().objectName() == "appRoot"
assert window.holdings_table.alternatingRowColors()
assert "QMainWindow" in window.styleSheet()
window.close()
service.close()
app.processEvents()
```
```python
def test_profit_and_loss_cells_use_a_share_colors(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from datetime import datetime
from decimal import Decimal
from PySide6.QtGui import QColor
from PySide6.QtWidgets import QApplication
from grid_trading.domain.models import PositionSummary
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
window._positions = [
PositionSummary(
instrument_id=1,
code="510300",
name="沪深300ETF",
total_quantity=100,
available_quantity=100,
base_quantity=0,
grid_quantity=100,
remaining_cost=Decimal("1000"),
average_cost=Decimal("10"),
position_breakeven_price=Decimal("10"),
account_breakeven_price=Decimal("10"),
realized_pnl=Decimal("12.34"),
grid_profit=Decimal("-5.67"),
current_price=Decimal("10.50"),
market_value=Decimal("1050"),
floating_pnl=Decimal("50"),
last_quote_at=datetime(2026, 7, 9, 10, 0, 0),
)
]
window._fill_holdings_table()
profit_item = window.holdings_table.item(0, 10)
loss_item = window.holdings_table.item(0, 11)
assert profit_item.foreground().color() == QColor("#c62828")
assert profit_item.font().bold()
assert loss_item.foreground().color() == QColor("#2e7d32")
assert loss_item.font().bold()
window.close()
service.close()
app.processEvents()
```
- [ ] **Step 2: Run the tests and verify they fail**
Run:
```bash
pytest tests/test_ui.py::test_main_window_applies_light_theme tests/test_ui.py::test_profit_and_loss_cells_use_a_share_colors -q
```
Expected: both new tests fail because object names, theme QSS, alternating row setup, and PnL item colors do not exist yet.
### Task 2: Add Light Theme and Shared Table Setup
**Files:**
- Modify: `src/grid_trading/ui/main_window.py`
- [ ] **Step 1: Add imports and theme constants**
Add `Decimal`, `QBrush`, `QColor`, `QFont`, and `QAbstractItemView` imports needed by the helpers.
- [ ] **Step 2: Apply theme in `_build_ui`**
Set `root.setObjectName("appRoot")`, call `self._apply_light_theme()`, and set object names on the navigation, summary card frames, toolbar buttons, tab widget, and tables.
- [ ] **Step 3: Add `_apply_light_theme` and `_configure_table`**
Create helpers that set the QSS theme once and apply consistent table settings: alternating rows, row selection, no editing, visible grid, row height, and header resize behavior.
- [ ] **Step 4: Run theme test**
Run:
```bash
pytest tests/test_ui.py::test_main_window_applies_light_theme -q
```
Expected: PASS.
### Task 3: Add PnL Styling Helpers
**Files:**
- Modify: `src/grid_trading/ui/main_window.py`
- [ ] **Step 1: Add `_style_pnl_item`, `_style_pnl_label`, and `_make_item`**
Use the raw `Decimal | None` value, not the formatted string, to decide color and font weight:
- `value > 0`: foreground `#c62828`, bold.
- `value < 0`: foreground `#2e7d32`, bold.
- otherwise: default foreground, non-bold.
- [ ] **Step 2: Use helpers in summary and tables**
Apply the helpers to summary floating PnL and these table columns:
- holdings: columns 10, 11, 12.
- grid levels: column 6.
- open grid lots: column 5.
- grid matches: column 7.
- [ ] **Step 3: Run PnL style test**
Run:
```bash
pytest tests/test_ui.py::test_profit_and_loss_cells_use_a_share_colors -q
```
Expected: PASS.
### Task 4: Full Verification
**Files:**
- No additional changes expected.
- [ ] **Step 1: Run UI test module**
Run:
```bash
pytest tests/test_ui.py -q
```
Expected: PASS.
- [ ] **Step 2: Run full test suite**
Run:
```bash
pytest -q
```
Expected: PASS.
- [ ] **Step 3: Review diff**
Run:
```bash
git diff -- src/grid_trading/ui/main_window.py tests/test_ui.py docs/superpowers/specs/2026-07-09-gui-visual-refresh-design.md docs/superpowers/plans/2026-07-09-gui-visual-refresh.md
```
Expected: diff only contains the UI theme, PnL styling, tests, and docs for this task.

View File

@@ -0,0 +1,464 @@
# Open Grid Lots Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Show unsold grid-buy lots with buy price, remaining quantity, suggested sell price, expected gross profit, current price, and sell readiness.
**Architecture:** Add a pure domain calculator that FIFO-matches grid sells against grid buys, expose it through `TradingService`, then render the result in a new PySide6 “待卖网格” table in the selected-instrument details area. The feature reuses existing trades, the default strategy template, and in-memory Tencent quote snapshots.
**Tech Stack:** Python 3.11, Decimal, PySide6, SQLite repository/service pattern, pytest.
---
## File Structure
- Create `src/grid_trading/domain/open_grid_lots.py`: pure FIFO calculation for unsold grid lots.
- Modify `src/grid_trading/domain/models.py`: add `OpenGridLot`.
- Modify `src/grid_trading/services/trading_service.py`: add `get_open_grid_lots(instrument_id, as_of=None)`.
- Modify `src/grid_trading/ui/main_window.py`: add the “待卖网格” table and refresh it with selected-instrument details.
- Create `tests/test_open_grid_lots.py`: domain FIFO and status tests.
- Modify `tests/test_services.py`: service integration test.
- Modify `tests/test_ui.py`: GUI smoke test.
- Modify `README.md`: document the new table.
## Tasks
### Task 1: Domain FIFO Calculator
**Files:**
- Create: `src/grid_trading/domain/open_grid_lots.py`
- Modify: `src/grid_trading/domain/models.py`
- Test: `tests/test_open_grid_lots.py`
- [ ] **Step 1: Write failing domain tests**
Add:
```python
from datetime import date, timedelta
from decimal import Decimal
from grid_trading.domain.models import Trade, TradeGroup, TradeSide
from grid_trading.domain.open_grid_lots import calculate_open_grid_lots
def make_trade(*, trade_id, trade_date, side, price, quantity, trade_group=TradeGroup.GRID):
return Trade(
id=trade_id,
account_id=1,
instrument_id=1,
trade_date=trade_date,
side=side,
price=Decimal(price),
quantity=quantity,
trade_group=trade_group,
)
def test_calculate_open_grid_lots_fifo_matches_grid_sells_against_grid_buys():
today = date(2026, 7, 9)
trades = [
make_trade(trade_id=1, trade_date=today - timedelta(days=3), side=TradeSide.BUY, price="10.00", quantity=300),
make_trade(trade_id=2, trade_date=today - timedelta(days=2), side=TradeSide.BUY, price="9.50", quantity=200),
make_trade(trade_id=3, trade_date=today - timedelta(days=1), side=TradeSide.SELL, price="10.30", quantity=350),
make_trade(
trade_id=4,
trade_date=today,
side=TradeSide.BUY,
price="8.00",
quantity=100,
trade_group=TradeGroup.BASE,
),
]
lots = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=Decimal("9.80"),
as_of=today,
)
assert len(lots) == 1
[lot] = lots
assert lot.buy_trade_id == 2
assert lot.buy_date == today - timedelta(days=2)
assert lot.buy_price == Decimal("9.50")
assert lot.remaining_quantity == 150
assert lot.actual_investment == Decimal("1425.00")
assert lot.suggested_sell_price == Decimal("9.79")
assert lot.estimated_gross_profit == Decimal("43.50")
assert lot.current_price == Decimal("9.80")
assert lot.status == "可卖"
def test_calculate_open_grid_lots_reports_not_reached_and_missing_quote_statuses():
today = date(2026, 7, 9)
trades = [
make_trade(trade_id=1, trade_date=today, side=TradeSide.BUY, price="10.00", quantity=100),
]
[not_reached] = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=Decimal("10.20"),
as_of=today,
)
[missing_quote] = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=None,
as_of=today,
)
assert not_reached.suggested_sell_price == Decimal("10.30")
assert not_reached.status == "未到价"
assert missing_quote.current_price is None
assert missing_quote.status == "未刷新行情"
```
- [ ] **Step 2: Run domain tests and verify red**
Run:
```powershell
pytest tests/test_open_grid_lots.py -q
```
Expected: FAIL because `grid_trading.domain.open_grid_lots` does not exist.
- [ ] **Step 3: Add model and implementation**
Add `OpenGridLot` to `models.py`:
```python
@dataclass(frozen=True)
class OpenGridLot:
buy_trade_id: int | None
buy_date: date
buy_price: Decimal
remaining_quantity: int
actual_investment: Decimal
suggested_sell_price: Decimal
estimated_gross_profit: Decimal
current_price: Decimal | None
status: str
```
Create `open_grid_lots.py` with FIFO matching:
```python
from __future__ import annotations
from dataclasses import replace
from datetime import date
from decimal import Decimal
from grid_trading.domain.calculations import money, price
from grid_trading.domain.models import OpenGridLot, Trade, TradeGroup, TradeSide
def calculate_open_grid_lots(
trades: list[Trade],
*,
spacing: Decimal,
current_price: Decimal | None,
as_of: date,
) -> list[OpenGridLot]:
if spacing <= 0 or spacing >= 1:
raise ValueError("网格间距必须大于 0 且小于 100%")
lots: list[OpenGridLot] = []
for trade in sorted(trades, key=lambda item: (item.trade_date, item.id or 0)):
if trade.trade_date > as_of or trade.trade_group is not TradeGroup.GRID:
continue
if trade.side is TradeSide.BUY:
suggested_sell_price = price(trade.price * (Decimal("1") + spacing))
lots.append(
OpenGridLot(
buy_trade_id=trade.id,
buy_date=trade.trade_date,
buy_price=price(trade.price),
remaining_quantity=trade.quantity,
actual_investment=money(trade.price * Decimal(trade.quantity)),
suggested_sell_price=suggested_sell_price,
estimated_gross_profit=money((suggested_sell_price - price(trade.price)) * Decimal(trade.quantity)),
current_price=current_price,
status=_status(current_price, suggested_sell_price),
)
)
else:
remaining_sell_quantity = trade.quantity
updated_lots: list[OpenGridLot] = []
for lot in lots:
if remaining_sell_quantity <= 0:
updated_lots.append(lot)
continue
matched_quantity = min(lot.remaining_quantity, remaining_sell_quantity)
remaining_sell_quantity -= matched_quantity
remaining_quantity = lot.remaining_quantity - matched_quantity
if remaining_quantity > 0:
updated_lots.append(_with_remaining_quantity(lot, remaining_quantity))
lots = updated_lots
return lots
def _with_remaining_quantity(lot: OpenGridLot, quantity: int) -> OpenGridLot:
return replace(
lot,
remaining_quantity=quantity,
actual_investment=money(lot.buy_price * Decimal(quantity)),
estimated_gross_profit=money((lot.suggested_sell_price - lot.buy_price) * Decimal(quantity)),
)
def _status(current_price: Decimal | None, suggested_sell_price: Decimal) -> str:
if current_price is None:
return "未刷新行情"
if current_price >= suggested_sell_price:
return "可卖"
return "未到价"
```
- [ ] **Step 4: Run domain tests and verify green**
Run:
```powershell
pytest tests/test_open_grid_lots.py -q
```
Expected: PASS.
### Task 2: Service API
**Files:**
- Modify: `src/grid_trading/services/trading_service.py`
- Test: `tests/test_services.py`
- [ ] **Step 1: Write failing service test**
Add:
```python
def test_service_returns_open_grid_lots_with_suggested_sell_price(tmp_path):
service = TradingService(tmp_path / "grid.db", quote_provider=FakeQuoteProvider())
service.ensure_defaults()
account = service.get_active_account()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("4.00"),
quantity=1000,
trade_group=TradeGroup.GRID,
)
)
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 8),
side=TradeSide.SELL,
price=Decimal("4.12"),
quantity=400,
trade_group=TradeGroup.GRID,
)
)
service.refresh_quotes()
[lot] = service.get_open_grid_lots(instrument.id, as_of=date(2026, 7, 9))
assert lot.buy_price == Decimal("4.00")
assert lot.remaining_quantity == 600
assert lot.suggested_sell_price == Decimal("4.12")
assert lot.current_price == Decimal("4.12")
assert lot.status == "可卖"
```
- [ ] **Step 2: Run service test and verify red**
Run:
```powershell
pytest tests/test_services.py::test_service_returns_open_grid_lots_with_suggested_sell_price -q
```
Expected: FAIL because `get_open_grid_lots` does not exist.
- [ ] **Step 3: Implement service method**
Import `calculate_open_grid_lots` and `OpenGridLot`, then add:
```python
def get_open_grid_lots(
self,
instrument_id: int,
*,
as_of: date | None = None,
) -> list[OpenGridLot]:
as_of_date = as_of or date.today()
self._require_instrument(instrument_id)
account = self.get_active_account()
trades = self.repository.list_trades(account_id=account.id, instrument_id=instrument_id)
position = next(
(
item
for item in self.get_position_summaries(as_of=as_of_date)
if item.instrument_id == instrument_id
),
None,
)
template = self.get_default_strategy_template()
return calculate_open_grid_lots(
trades,
spacing=template.grid_spacing_pct,
current_price=position.current_price if position is not None else None,
as_of=as_of_date,
)
```
- [ ] **Step 4: Run service tests and verify green**
Run:
```powershell
pytest tests/test_services.py -q
```
Expected: PASS.
### Task 3: GUI Table
**Files:**
- Modify: `src/grid_trading/ui/main_window.py`
- Test: `tests/test_ui.py`
- [ ] **Step 1: Write failing GUI smoke test**
Add:
```python
def test_main_window_contains_open_grid_lots_table(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QGroupBox
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.open_grid_lots_table.columnCount() == 8
assert any(group.title() == "待卖网格" for group in window.findChildren(QGroupBox))
window.close()
service.close()
app.processEvents()
```
- [ ] **Step 2: Run GUI test and verify red**
Run:
```powershell
pytest tests/test_ui.py::test_main_window_contains_open_grid_lots_table -q
```
Expected: FAIL because `open_grid_lots_table` does not exist.
- [ ] **Step 3: Implement GUI table**
Add `OPEN_GRID_LOT_COLUMNS`:
```python
OPEN_GRID_LOT_COLUMNS = [
"买入日期",
"买入价",
"剩余股数",
"实际投入",
"建议卖出价",
"预计毛利润",
"当前价",
"状态",
]
```
Create `self.open_grid_lots_table` in `_build_ui`, add a `QGroupBox("待卖网格")`, and refresh it from `_refresh_details`:
```python
def _refresh_open_grid_lots(self, position: PositionSummary | None) -> None:
if position is None:
self._fill_open_grid_lots_table([])
return
lots = self.service.get_open_grid_lots(position.instrument_id)
self._fill_open_grid_lots_table(lots)
def _fill_open_grid_lots_table(self, lots) -> None:
self.open_grid_lots_table.setRowCount(len(lots))
for row, lot in enumerate(lots):
values = [
lot.buy_date.isoformat(),
format_price(lot.buy_price),
format_quantity(lot.remaining_quantity),
format_money(lot.actual_investment),
format_price(lot.suggested_sell_price),
format_money(lot.estimated_gross_profit),
format_price(lot.current_price),
lot.status,
]
for column, value in enumerate(values):
self.open_grid_lots_table.setItem(row, column, QTableWidgetItem(value))
```
- [ ] **Step 4: Run GUI tests and verify green**
Run:
```powershell
pytest tests/test_ui.py -q
```
Expected: PASS.
### Task 4: Docs And Verification
**Files:**
- Modify: `README.md`
- [ ] **Step 1: Update README**
Document that the selected-instrument area contains a “待卖网格” table showing unsold grid buys and suggested sell prices.
- [ ] **Step 2: Run all tests**
Run:
```powershell
pytest -q
```
Expected: all tests pass.
- [ ] **Step 3: Run CLI smoke**
Run:
```powershell
python -m grid_trading.app --help
```
Expected: help text prints normally.
## Self-Review
- Spec coverage: FIFO matching, suggested sell price, expected profit, current-price status, service API, GUI table, and README are covered.
- Placeholder scan: no unresolved placeholder text is intentionally left.
- Type consistency: `OpenGridLot`, `calculate_open_grid_lots`, and `get_open_grid_lots` names match across tasks.

View File

@@ -13,7 +13,7 @@
- 应用形态Python 桌面 GUI。
- 技术路线PySide6 + SQLite + 分层业务计算模块。
- 交易市场A股股票和 ETF。
- 数据来源:成交手动录入;行情后续半自动刷新。
- 数据来源:成交手动录入;现价通过腾讯接口手动刷新。
- 首屏布局:持仓表优先,左侧导航,顶部账户摘要,中间持仓表,下方选中标的详情。
- 网格策略:默认模板 + 单标的覆盖。
- 回本价:同时显示持仓回本价和账户回本价。
@@ -40,14 +40,14 @@
4. 成交记录
- 手动录入买入、卖出成交。
- 成交字段包括日期、代码、方向、价、数量、手续费、印花税、过户费、交易分组、备注。
- 成交字段包括日期、代码、方向、成交价、数量、手续费、印花税、过户费、交易分组、备注。
- 交易分组包括底仓、网格、其他,用于区分底仓和网格仓。
- 手续费支持自动估算和手动覆盖,费率在设置中可配置。
- 支持修改和删除成交,保存后自动重算相关持仓。
5. 持仓表
- 每个标的一行,展示代码、名称、当前价格、总持仓、可用数量、底仓数量、网格仓数量、持仓成本、持仓回本价、账户回本价、已实现盈亏、累计网格利润、浮动盈亏。
- 第一阶段没有行情源时,当前价格可手动维护;未维护价格时使用最近成交价作为估值参考,并在界面中标记
- 当前价格只来自腾讯行情刷新;没有行情快照时显示空值,不用成交价或标的字段兜底估值
- A股 T+1 可用数量按交易日期计算:当日买入数量不可卖出,当日卖出会减少可用数量。
6. 数据持久化
@@ -61,7 +61,7 @@
- 不自动登录券商。
- 不自动下单。
-接实时行情。
-自动定时刷新行情。
- 不生成盘中弹窗提醒。
- 不做复杂图表。
- 不做多账户同步或云端备份。
@@ -85,7 +85,7 @@ GUI 只负责展示和收集输入;所有计算通过 service 调用 domain
核心表:
- `accounts`:账户基础信息、初始现金、当前现金。
- `instruments`:标的信息,包括代码、名称、市场、交易单位、手动价格
- `instruments`:标的信息,包括代码、名称、市场、交易单位。
- `strategy_templates`:默认网格模板。
- `instrument_strategy_overrides`:单标的策略覆盖。
- `trades`:成交记录。
@@ -125,7 +125,7 @@ GUI 只负责展示和收集输入;所有计算通过 service 调用 domain
估值:
- 持仓市值 = 当前价格 * 当前总持仓数量。
- 当前价格优先使用手动维护价格;没有手动价格时使用最近成交价,并显示“估值价来自最近成交”
- 当前价格只使用腾讯接口返回的行情快照;没有行情快照时显示为空,持仓市值和浮动盈亏也显示为空
- 浮动盈亏 = 持仓市值 - 当前剩余持仓成本。
## 主界面
@@ -138,7 +138,7 @@ GUI 只负责展示和收集输入;所有计算通过 service 调用 domain
- 中央持仓表:展示所有标的的核心指标。
- 底部详情区:展示选中标的的最近成交、分组持仓、策略参数和计算说明。
第一阶段的“刷新”只从数据库重算,不调用行情接口
“刷新行情”从腾讯接口拉取当前价格并重算持仓表和账户摘要
## 错误处理
@@ -172,7 +172,7 @@ GUI 只负责展示和收集输入;所有计算通过 service 调用 domain
## 后续阶段
第二阶段:接入 A股/ETF 行情源,支持手动刷新和定时刷新,生成买卖档位和今日委托建议。
第二阶段:支持定时刷新行情,生成买卖档位和今日委托建议。
第三阶段:加入图表,包括成本变化曲线、累计网格收益、账户盈亏和资金使用率。

View File

@@ -6,7 +6,7 @@ Date: 2026-07-08
Add Tencent A-share quote integration so the GUI holdings table displays realtime prices from `http://qt.gtimg.cn/q=<symbol>`.
The first implementation is manual refresh only. It does not auto-refresh, generate buy/sell grid orders, or overwrite user-entered manual prices.
The first implementation is manual refresh only. It does not auto-refresh, generate buy/sell grid orders, persist quote prices, or use trade-entry prices as current prices.
## Confirmed Decision
@@ -45,14 +45,12 @@ Add a `market` package with a Tencent quote client. The client converts local in
`TradingService` keeps an in-memory quote cache. When the user clicks refresh in the GUI, the service fetches quotes for all active instruments and stores them in memory.
Position calculations will prefer prices in this order:
Position calculations will treat Tencent quotes as the only current-price source:
1. Tencent realtime quote
2. Manual price
3. Last trade price
4. Empty value
2. Empty value
The holdings table and account summary will immediately reflect quote prices after refresh. Quote prices are not persisted to SQLite in this step, so stale prices do not silently survive app restarts.
The holdings table and account summary will immediately reflect quote prices after refresh. Quote prices are not persisted to SQLite in this step, so stale prices do not silently survive app restarts. Trade-entry prices remain historical成交价 only and never become current-price fallbacks.
## Error Handling
@@ -65,6 +63,6 @@ The holdings table and account summary will immediately reflect quote prices aft
- Unit-test Tencent response parsing with a fixture response.
- Unit-test symbol inference.
- Unit-test position calculations with a realtime quote overriding manual price.
- Unit-test position calculations with a realtime quote supplying current price and a missing quote leaving current price empty.
- Unit-test service quote refresh using a fake quote provider.
- Keep existing GUI smoke tests passing.

View File

@@ -0,0 +1,106 @@
# 网格档位建议设计
日期2026-07-09
## 目标
在 GUI 中为选中的 A股/ETF 标的生成网格档位表,帮助用户根据腾讯接口刷新的现价和现有策略模板,快速看到每一档的买入价、建议股数、实际投入、卖出价和预计单轮毛利润。
该功能只做计算和展示,不自动下单,不自动创建成交记录,也不保存生成结果。
## 已确认口径
- 买入档位使用现有策略模板里的 `网格间距`,按百分比逐档递减。
- 卖出价也使用同一个 `网格间距`,即买入后上涨同样百分比作为该档卖出价。
- 买入金额使用策略模板里的 `每格金额`
- 现价只来自腾讯行情快照;没有现价时不生成档位,并提示先刷新行情。
- 建议股数按标的交易单位向下取整,默认 100 股一手。
- 预计单轮毛利润不扣手续费,先保持和用户截图一致。
## 计算公式
设:
- `current_price`:腾讯行情现价
- `spacing`:策略模板网格间距,例如 3% = `0.03`
- `amount_per_grid`:策略模板每格金额
- `lot_size`:标的交易单位
- `level`:档位,从 1 开始
逐档计算:
```text
第1档买入价 = current_price * (1 - spacing)
第N档买入价 = 第N-1档买入价 * (1 - spacing)
建议买入股数 = floor(amount_per_grid / 买入价 / lot_size) * lot_size
实际投入 = 买入价 * 建议买入股数
卖出价 = 买入价 * (1 + spacing)
预计单轮毛利润 = (卖出价 - 买入价) * 建议买入股数
```
价格按现有 `price()` 规则保留两位小数,金额按现有 `money()` 规则保留两位小数。
如果金额不足以买入一个交易单位,该档股数为 0实际投入和预计利润为 0。
## 界面设计
在主窗口下方选中标的区域新增“网格档位”表,列为:
```text
档位 / 买入价 / 买入金额 / 建议买入股数 / 实际投入 / 卖出价 / 预计单轮毛利润
```
表格随选中标的变化自动刷新。点击“刷新行情”后,持仓表和网格档位表一起刷新。
第一版提供一个非持久化档数输入,默认 10 档。用户改档数后即时重算,不写入 SQLite。
## 代码结构
新增独立的领域计算模块:
- `src/grid_trading/domain/grid_levels.py`
- `generate_grid_levels(...)`
- 只依赖 Decimal 和领域模型,便于单元测试。
扩展领域模型:
- `GridLevelSuggestion`
- `level`
- `buy_price`
- `buy_amount`
- `suggested_quantity`
- `actual_investment`
- `sell_price`
- `estimated_gross_profit`
扩展服务层:
- `TradingService.get_grid_level_suggestions(instrument_id, levels=10)`
- 获取当前持仓摘要中的现价。
- 获取默认策略模板。
- 获取标的交易单位。
- 没有现价时返回空列表,由 UI 显示提示。
## 错误处理
- 没有选中标的:档位表清空。
- 没有腾讯现价:档位表清空并显示“请先刷新行情”。
- 网格间距、每格金额或交易单位无效服务层抛出可读错误UI 弹窗提示。
- 档数最小为 1最大为 100避免生成过多行影响界面。
## 测试策略
- 单元测试覆盖百分比递减买入价。
- 单元测试覆盖建议股数按交易单位向下取整。
- 单元测试覆盖卖出价和预计单轮毛利润。
- 服务测试覆盖有现价时生成档位、无现价时返回空列表。
- GUI 冒烟测试覆盖“网格档位”表存在,并能在 offscreen 模式构造。
## 非目标
- 不生成真实委托。
- 不自动写入成交记录。
- 不扣除手续费计算净利润。
- 不做单标的策略覆盖编辑。
- 不持久化每次生成的档位表。

View File

@@ -0,0 +1,53 @@
# GUI 页面美化与盈亏着色设计
日期2026-07-09
## 目标
把现有 PySide6 桌面 GUI 调整为清爽浅色桌面工具风格,并统一处理盈亏、利润字段的视觉反馈:亏损显示绿色加粗,盈利显示红色加粗,零值和空值保持中性显示。
本次只改视觉表现和表格单元格样式,不改变业务计算、数据结构、行情刷新、成交录入或持仓选择逻辑。
## 视觉方向
采用轻量主题化方案:
- 保留当前左侧导航、顶部摘要、工具栏、持仓表和底部详情 tab 的布局结构。
- 主窗口使用浅灰背景,内容区使用白色和浅边框分区,减少默认 Qt 控件的粗糙感。
- 侧边栏使用白底、蓝色选中态和更紧凑的列表项。
- 摘要卡片使用白底、浅边框、圆角和更清晰的标题/数值层级。
- 操作按钮使用蓝色主按钮样式,禁用态降低对比。
- 表格使用白底、浅灰表头、斑马纹、整行选中和统一网格线颜色。
- Tab 使用浅色选项卡,当前 tab 用蓝色强调。
## 盈亏字段
需要应用盈亏样式的字段:
- 汇总卡:浮动盈亏。
- 持仓表:已实现盈亏、网格利润、浮动盈亏。
- 网格档位表:预计单轮毛利润。
- 待卖网格表:预计毛利润。
- 配对明细表:毛利润。
样式规则:
- 大于 0红色、加粗。
- 小于 0绿色、加粗。
- 等于 0、空值或无法解析默认文字色和普通字重。
颜色按 A 股常见习惯处理,红色代表盈利,绿色代表亏损。
## 实现方式
-`MainWindow` 中新增一个全局样式应用方法,用 QSS 统一主窗口、按钮、列表、表格、tab 和摘要卡片样式。
- 给摘要卡片和关键控件设置 objectName让 QSS 精准命中,避免影响弹窗内部布局。
- 新增表格初始化 helper复用表头、选择、斑马纹、行高和滚动行为设置。
- 新增数值单元格 helper根据 Decimal 原始值给 `QTableWidgetItem` 设置前景色和加粗字体。
- 刷新汇总卡时对浮动盈亏标签应用同样的盈亏样式。
## 测试策略
- 增加 GUI 烟测,验证主窗口应用了浅色主题相关 objectName 或样式入口。
- 增加单元格样式测试,直接填充带盈利和亏损的持仓数据,验证盈利字段为红色加粗、亏损字段为绿色加粗。
- 保留现有窗口构造、tab、刷新线程测试。

View File

@@ -0,0 +1,46 @@
# 待卖网格明细设计
日期2026-07-09
## 目标
在选中标的详情区展示“待卖网格”明细,让用户看清当前网格仓由哪些尚未卖出的网格买入组成、每笔买入价是多少、按策略应该挂到什么卖出价。
该功能只做计算和展示,不自动下单,不自动生成成交。
## 已确认口径
- 只统计成交分组为 `网格` 的成交。
- 网格买入形成一笔待卖批次。
- 网格卖出按时间顺序 FIFO 抵消最早的待卖批次。
- 仍有剩余数量的买入批次显示在“待卖网格”表中。
- 建议卖出价 = 买入价 × (1 + 默认策略模板的网格间距)。
- 预计毛利润 = (建议卖出价 - 买入价) × 剩余股数,不扣手续费。
- 如果已经刷新行情,则显示当前价,并用当前价判断状态:
- 当前价 >= 建议卖出价:`可卖`
- 当前价 < 建议卖出价`未到价`
- 没有现价`未刷新行情`
## 展示列
新增表格标题`待卖网格`
```text
买入日期 / 买入价 / 剩余股数 / 实际投入 / 建议卖出价 / 预计毛利润 / 当前价 / 状态
```
## 代码结构
- 新增领域模型 `OpenGridLot`表示一笔尚未卖完的网格买入
- 新增领域计算模块 `open_grid_lots.py`根据成交记录网格间距和现价计算待卖批次
- 新增服务方法 `TradingService.get_open_grid_lots(instrument_id)` GUI 调用
- 主窗口下方详情区域新增待卖网格选中标的变化刷新行情录入/编辑/删除成交后自动刷新
## 非目标
- 不改变现有持仓成本和网格利润算法
- 不自动匹配券商委托
- 不新增数据库表
- 不扣除手续费估算净利润

View File

@@ -75,7 +75,6 @@ def calculate_positions(
lambda: {group: _GroupState() for group in TradeGroup}
)
net_invested: dict[int, Decimal] = defaultdict(lambda: Decimal("0"))
last_trade_price: dict[int, Decimal] = {}
today_buys: dict[int, int] = defaultdict(int)
for trade in sorted(trades, key=lambda item: (item.trade_date, item.id or 0)):
@@ -89,8 +88,6 @@ def calculate_positions(
group_state = states[trade.instrument_id][trade.trade_group]
gross = money(trade.gross_amount)
fees = money(trade.total_fee)
last_trade_price[trade.instrument_id] = trade.price
if trade.side is TradeSide.BUY:
group_state.quantity += trade.quantity
group_state.cost = money(group_state.cost + gross + fees)
@@ -123,11 +120,7 @@ def calculate_positions(
remaining_cost = money(sum((state.cost for state in group_states.values()), Decimal("0")))
realized_pnl = money(sum((state.realized_pnl for state in group_states.values()), Decimal("0")))
grid_profit = money(group_states[TradeGroup.GRID].realized_pnl)
current_price, price_source = _resolve_current_price(
instrument,
last_trade_price.get(instrument.id),
quotes.get(instrument.id),
)
current_price, price_source = _resolve_current_price(quotes.get(instrument.id))
market_value = money(current_price * Decimal(total_quantity)) if current_price is not None else None
floating_pnl = money(market_value - remaining_cost) if market_value is not None else None
available_quantity = max(0, total_quantity - today_buys[instrument.id])
@@ -184,9 +177,10 @@ def calculate_account_summary(
sum((position.floating_pnl or Decimal("0") for position in positions), Decimal("0"))
)
total_assets = money(cash + market_value)
deployable_assets = market_value + max(cash, Decimal("0"))
capital_usage_rate = (
(market_value / total_assets).quantize(RATE_PLACES, rounding=ROUND_HALF_UP)
if total_assets > 0
(market_value / deployable_assets).quantize(RATE_PLACES, rounding=ROUND_HALF_UP)
if deployable_assets > 0
else Decimal("0")
)
return AccountSummary(
@@ -207,17 +201,9 @@ def _validate_trade(trade: Trade) -> None:
raise CalculationError("Trade fees cannot be negative")
def _resolve_current_price(
instrument: Instrument,
fallback_trade_price: Decimal | None,
quote_snapshot: QuoteSnapshot | None,
) -> tuple[Decimal | None, str]:
def _resolve_current_price(quote_snapshot: QuoteSnapshot | None) -> tuple[Decimal | None, str]:
if quote_snapshot is not None:
return quote_snapshot.price, quote_snapshot.source
if instrument.manual_price is not None:
return instrument.manual_price, "manual"
if fallback_trade_price is not None:
return fallback_trade_price, "last_trade"
return None, "missing"

View File

@@ -0,0 +1,47 @@
from __future__ import annotations
from decimal import Decimal
from grid_trading.domain.calculations import money, price
from grid_trading.domain.models import GridLevelSuggestion
def generate_grid_levels(
*,
current_price: Decimal,
spacing: Decimal,
amount_per_grid: Decimal,
lot_size: int,
levels: int,
) -> list[GridLevelSuggestion]:
if current_price <= 0:
raise ValueError("现价必须大于 0")
if spacing <= 0 or spacing >= 1:
raise ValueError("网格间距必须大于 0 且小于 100%")
if amount_per_grid <= 0:
raise ValueError("每格金额必须大于 0")
if lot_size <= 0:
raise ValueError("交易单位必须大于 0")
if levels < 1 or levels > 100:
raise ValueError("档数必须在 1 到 100 之间")
suggestions: list[GridLevelSuggestion] = []
buy_price = price(current_price * (Decimal("1") - spacing))
for level in range(1, levels + 1):
quantity = int(amount_per_grid / buy_price) // lot_size * lot_size
actual_investment = money(buy_price * Decimal(quantity))
sell_price = price(buy_price * (Decimal("1") + spacing))
estimated_gross_profit = money((sell_price - buy_price) * Decimal(quantity))
suggestions.append(
GridLevelSuggestion(
level=level,
buy_price=buy_price,
buy_amount=money(amount_per_grid),
suggested_quantity=quantity,
actual_investment=actual_investment,
sell_price=sell_price,
estimated_gross_profit=estimated_gross_profit,
)
)
buy_price = price(buy_price * (Decimal("1") - spacing))
return suggestions

View File

@@ -114,6 +114,44 @@ class FeeEstimate:
return self.commission + self.stamp_tax + self.transfer_fee
@dataclass(frozen=True)
class GridLevelSuggestion:
level: int
buy_price: Decimal
buy_amount: Decimal
suggested_quantity: int
actual_investment: Decimal
sell_price: Decimal
estimated_gross_profit: Decimal
@dataclass(frozen=True)
class OpenGridLot:
buy_trade_id: int | None
buy_date: date
buy_price: Decimal
remaining_quantity: int
actual_investment: Decimal
suggested_sell_price: Decimal
estimated_gross_profit: Decimal
current_price: Decimal | None
status: str
@dataclass(frozen=True)
class GridTradeMatch:
sell_trade_id: int | None
sell_date: date
sell_price: Decimal
buy_trade_id: int | None
buy_date: date
buy_price: Decimal
matched_quantity: int
buy_amount: Decimal
sell_amount: Decimal
gross_profit: Decimal
@dataclass(frozen=True)
class QuoteSnapshot:
symbol: str

View File

@@ -0,0 +1,131 @@
from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import date
from decimal import Decimal
from grid_trading.domain.calculations import money, price
from grid_trading.domain.models import GridTradeMatch, OpenGridLot, Trade, TradeGroup, TradeSide
@dataclass
class _OpenBuyLot:
trade_id: int | None
buy_date: date
buy_price: Decimal
remaining_quantity: int
def calculate_open_grid_lots(
trades: list[Trade],
*,
spacing: Decimal,
current_price: Decimal | None,
as_of: date,
) -> list[OpenGridLot]:
if spacing <= 0 or spacing >= 1:
raise ValueError("网格间距必须大于 0 且小于 100%")
lots: list[OpenGridLot] = []
for trade in sorted(trades, key=lambda item: (item.trade_date, item.id or 0)):
if trade.trade_date > as_of or trade.trade_group is not TradeGroup.GRID:
continue
if trade.side is TradeSide.BUY:
buy_price = price(trade.price)
suggested_sell_price = price(buy_price * (Decimal("1") + spacing))
lots.append(
OpenGridLot(
buy_trade_id=trade.id,
buy_date=trade.trade_date,
buy_price=buy_price,
remaining_quantity=trade.quantity,
actual_investment=money(buy_price * Decimal(trade.quantity)),
suggested_sell_price=suggested_sell_price,
estimated_gross_profit=money(
(suggested_sell_price - buy_price) * Decimal(trade.quantity)
),
current_price=current_price,
status=_status(current_price, suggested_sell_price),
)
)
else:
lots = _match_sell_against_lots(lots, trade.quantity)
return lots
def calculate_grid_trade_matches(
trades: list[Trade],
*,
as_of: date,
) -> list[GridTradeMatch]:
open_buys: list[_OpenBuyLot] = []
matches: list[GridTradeMatch] = []
for trade in sorted(trades, key=lambda item: (item.trade_date, item.id or 0)):
if trade.trade_date > as_of or trade.trade_group is not TradeGroup.GRID:
continue
if trade.side is TradeSide.BUY:
open_buys.append(
_OpenBuyLot(
trade_id=trade.id,
buy_date=trade.trade_date,
buy_price=price(trade.price),
remaining_quantity=trade.quantity,
)
)
continue
sell_price = price(trade.price)
remaining_sell_quantity = trade.quantity
while remaining_sell_quantity > 0 and open_buys:
buy_lot = open_buys[0]
matched_quantity = min(buy_lot.remaining_quantity, remaining_sell_quantity)
matches.append(
GridTradeMatch(
sell_trade_id=trade.id,
sell_date=trade.trade_date,
sell_price=sell_price,
buy_trade_id=buy_lot.trade_id,
buy_date=buy_lot.buy_date,
buy_price=buy_lot.buy_price,
matched_quantity=matched_quantity,
buy_amount=money(buy_lot.buy_price * Decimal(matched_quantity)),
sell_amount=money(sell_price * Decimal(matched_quantity)),
gross_profit=money((sell_price - buy_lot.buy_price) * Decimal(matched_quantity)),
)
)
buy_lot.remaining_quantity -= matched_quantity
remaining_sell_quantity -= matched_quantity
if buy_lot.remaining_quantity == 0:
open_buys.pop(0)
return matches
def _match_sell_against_lots(lots: list[OpenGridLot], quantity: int) -> list[OpenGridLot]:
remaining_sell_quantity = quantity
updated_lots: list[OpenGridLot] = []
for lot in lots:
if remaining_sell_quantity <= 0:
updated_lots.append(lot)
continue
matched_quantity = min(lot.remaining_quantity, remaining_sell_quantity)
remaining_sell_quantity -= matched_quantity
remaining_quantity = lot.remaining_quantity - matched_quantity
if remaining_quantity > 0:
updated_lots.append(_with_remaining_quantity(lot, remaining_quantity))
return updated_lots
def _with_remaining_quantity(lot: OpenGridLot, quantity: int) -> OpenGridLot:
return replace(
lot,
remaining_quantity=quantity,
actual_investment=money(lot.buy_price * Decimal(quantity)),
estimated_gross_profit=money((lot.suggested_sell_price - lot.buy_price) * Decimal(quantity)),
)
def _status(current_price: Decimal | None, suggested_sell_price: Decimal) -> str:
if current_price is None:
return "未刷新行情"
if current_price >= suggested_sell_price:
return "可卖"
return "未到价"

View File

@@ -50,10 +50,12 @@ class TencentQuoteProvider:
*,
endpoint: str = TENCENT_ENDPOINT,
timeout: float = 5.0,
retries: int = 1,
fetcher: Callable[[str, float], bytes] | None = None,
):
self.endpoint = endpoint
self.timeout = timeout
self.retries = retries
self._fetcher = fetcher or _default_fetcher
def fetch_quotes(self, instruments: Iterable[Instrument]) -> dict[str, QuoteSnapshot]:
@@ -62,12 +64,18 @@ class TencentQuoteProvider:
return {}
query = ",".join(symbols)
url = f"{self.endpoint}{query}"
try:
raw = self._fetcher(url, self.timeout)
except (OSError, HTTPError, URLError) as exc:
raise QuoteFetchError(f"Tencent quote request failed: {exc}") from exc
raw = self._fetch_with_retry(url)
return {quote.code: quote for quote in parse_tencent_response(raw)}
def _fetch_with_retry(self, url: str) -> bytes:
last_error: Exception | None = None
for _ in range(self.retries + 1):
try:
return self._fetcher(url, self.timeout)
except (OSError, HTTPError, URLError) as exc:
last_error = exc
raise QuoteFetchError(f"Tencent quote request failed: {last_error}") from last_error
def _default_fetcher(url: str, timeout: float) -> bytes:
with urlopen(url, timeout=timeout) as response:

View File

@@ -12,12 +12,16 @@ from grid_trading.domain.calculations import (
calculate_positions,
estimate_fees,
)
from grid_trading.domain.grid_levels import generate_grid_levels
from grid_trading.domain.models import (
Account,
AccountSummary,
FeeEstimate,
FeeRules,
GridLevelSuggestion,
GridTradeMatch,
Instrument,
OpenGridLot,
PositionSummary,
QuoteSnapshot,
StrategyOverride,
@@ -25,6 +29,7 @@ from grid_trading.domain.models import (
Trade,
TradeSide,
)
from grid_trading.domain.open_grid_lots import calculate_grid_trade_matches, calculate_open_grid_lots
from grid_trading.market.tencent import TencentQuoteProvider
from grid_trading.storage.repositories import Repository
@@ -94,10 +99,19 @@ class TradingService:
def refresh_quotes(self) -> dict[str, QuoteSnapshot]:
instruments = self.repository.list_instruments()
quotes = self.quote_provider.fetch_quotes(instruments)
self._quote_cache = quotes
quotes = self.fetch_quotes_for_instruments(instruments)
self.apply_quote_snapshots(quotes)
return quotes
def fetch_quotes_for_instruments(
self,
instruments: list[Instrument],
) -> dict[str, QuoteSnapshot]:
return self.quote_provider.fetch_quotes(instruments)
def apply_quote_snapshots(self, quotes: dict[str, QuoteSnapshot]) -> None:
self._quote_cache = quotes
def get_default_strategy_template(self) -> StrategyTemplate:
self.ensure_defaults()
template = self.repository.get_default_strategy_template()
@@ -197,6 +211,70 @@ class TradingService:
]
return calculate_account_summary(account, positions, ledger_entries, trades)
def get_grid_level_suggestions(
self,
instrument_id: int,
*,
levels: int = 10,
) -> list[GridLevelSuggestion]:
instrument = self._require_instrument(instrument_id)
position = next(
(
item
for item in self.get_position_summaries()
if item.instrument_id == instrument_id
),
None,
)
if position is None or position.current_price is None:
return []
template = self.get_default_strategy_template()
return generate_grid_levels(
current_price=position.current_price,
spacing=template.grid_spacing_pct,
amount_per_grid=template.amount_per_grid,
lot_size=instrument.lot_size,
levels=levels,
)
def get_open_grid_lots(
self,
instrument_id: int,
*,
as_of: date | None = None,
) -> list[OpenGridLot]:
as_of_date = as_of or date.today()
self._require_instrument(instrument_id)
account = self.get_active_account()
trades = self.repository.list_trades(account_id=account.id, instrument_id=instrument_id)
position = next(
(
item
for item in self.get_position_summaries(as_of=as_of_date)
if item.instrument_id == instrument_id
),
None,
)
template = self.get_default_strategy_template()
return calculate_open_grid_lots(
trades,
spacing=template.grid_spacing_pct,
current_price=position.current_price if position is not None else None,
as_of=as_of_date,
)
def get_grid_trade_matches(
self,
instrument_id: int,
*,
as_of: date | None = None,
) -> list[GridTradeMatch]:
as_of_date = as_of or date.today()
self._require_instrument(instrument_id)
account = self.get_active_account()
trades = self.repository.list_trades(account_id=account.id, instrument_id=instrument_id)
return calculate_grid_trade_matches(trades, as_of=as_of_date)
def _require_account(self, account_id: int) -> Account:
account = self.repository.get_account(account_id)
if account is None:

View File

@@ -1,7 +1,7 @@
from __future__ import annotations
from datetime import date
from decimal import Decimal, InvalidOperation
from decimal import Decimal
from PySide6.QtCore import QDate
from PySide6.QtWidgets import (
@@ -81,7 +81,6 @@ class InstrumentDialog(QDialog):
self.lot_size_edit = QSpinBox()
self.lot_size_edit.setRange(1, 1_000_000)
self.lot_size_edit.setValue(instrument.lot_size if instrument else 100)
self.manual_price_edit = QLineEdit(str(instrument.manual_price or "") if instrument else "")
self.allow_odd_lot_edit = QCheckBox("允许非整手")
self.allow_odd_lot_edit.setChecked(instrument.allow_odd_lot if instrument else False)
@@ -90,7 +89,6 @@ class InstrumentDialog(QDialog):
form.addRow("名称", self.name_edit)
form.addRow("市场", self.market_edit)
form.addRow("交易单位", self.lot_size_edit)
form.addRow("手动价格", self.manual_price_edit)
form.addRow("", self.allow_odd_lot_edit)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
@@ -102,14 +100,13 @@ class InstrumentDialog(QDialog):
layout.addWidget(buttons)
def to_instrument(self) -> Instrument:
manual_price = _optional_decimal(self.manual_price_edit.text().strip(), "手动价格")
return Instrument(
id=self._instrument.id if self._instrument else None,
code=self.code_edit.text().strip(),
name=self.name_edit.text().strip(),
market=self.market_edit.currentText(),
lot_size=self.lot_size_edit.value(),
manual_price=manual_price,
manual_price=None,
allow_odd_lot=self.allow_odd_lot_edit.isChecked(),
active=True,
)
@@ -227,7 +224,7 @@ class TradeDialog(QDialog):
form.addRow("日期", self.date_edit)
form.addRow("方向", self.side_edit)
form.addRow("分组", self.group_edit)
form.addRow("", self.price_edit)
form.addRow("成交", self.price_edit)
form.addRow("数量", self.quantity_edit)
form.addRow(fee_row)
form.addRow("备注", self.notes_edit)
@@ -291,10 +288,3 @@ def _percent_spinbox(value: Decimal) -> QDoubleSpinBox:
return spinbox
def _optional_decimal(raw: str, label: str) -> Decimal | None:
if not raw:
return None
try:
return Decimal(raw)
except InvalidOperation as exc:
raise ValueError(f"{label} 不是有效数字") from exc

View File

@@ -1,15 +1,17 @@
from __future__ import annotations
import sys
from decimal import Decimal
from pathlib import Path
from PySide6.QtCore import Qt
from PySide6.QtCore import QObject, Qt, QThread, Signal
from PySide6.QtGui import QBrush, QColor, QFont
from PySide6.QtWidgets import (
QAbstractItemView,
QApplication,
QDialog,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QHeaderView,
QLabel,
@@ -18,6 +20,8 @@ from PySide6.QtWidgets import (
QMessageBox,
QPushButton,
QSplitter,
QSpinBox,
QTabWidget,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
@@ -25,13 +29,42 @@ from PySide6.QtWidgets import (
)
from grid_trading.config import DEFAULT_DB_PATH
from grid_trading.domain.models import PositionSummary, Trade, TradeSide
from grid_trading.domain.models import (
GridLevelSuggestion,
GridTradeMatch,
OpenGridLot,
PositionSummary,
Trade,
TradeSide,
)
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.dialogs import AccountDialog, InstrumentDialog, StrategyTemplateDialog, TradeDialog
from grid_trading.ui.formatters import format_money, format_percent, format_price, format_quantity
class QuoteRefreshWorker(QObject):
finished = Signal(object)
failed = Signal(str)
def __init__(self, service: TradingService, instruments):
super().__init__()
self._service = service
self._instruments = instruments
def run(self) -> None:
try:
quotes = self._service.fetch_quotes_for_instruments(self._instruments)
except Exception as exc:
self.failed.emit(str(exc))
return
self.finished.emit(quotes)
class MainWindow(QMainWindow):
PROFIT_COLOR = QColor("#c62828")
LOSS_COLOR = QColor("#2e7d32")
NEUTRAL_COLOR = QColor("#334155")
HOLDING_COLUMNS = [
"代码",
"名称",
@@ -48,6 +81,36 @@ class MainWindow(QMainWindow):
"浮动盈亏",
]
TRADE_COLUMNS = ["日期", "方向", "分组", "价格", "数量", "费用", "备注"]
GRID_LEVEL_COLUMNS = [
"档位",
"买入价",
"买入金额",
"建议买入股数",
"实际投入",
"卖出价",
"预计单轮毛利润",
]
OPEN_GRID_LOT_COLUMNS = [
"买入日期",
"买入价",
"剩余股数",
"实际投入",
"建议卖出价",
"预计毛利润",
"当前价",
"状态",
]
GRID_MATCH_COLUMNS = [
"卖出日期",
"卖出价",
"匹配买入日期",
"买入价",
"匹配股数",
"买入金额",
"卖出金额",
"毛利润",
"对应成交",
]
def __init__(self, service: TradingService):
super().__init__()
@@ -56,6 +119,9 @@ class MainWindow(QMainWindow):
self._positions: list[PositionSummary] = []
self._selected_instrument_id: int | None = None
self._trade_ids_by_row: dict[int, int] = {}
self._quote_thread: QThread | None = None
self._quote_worker: QuoteRefreshWorker | None = None
self.refresh_quotes_button: QPushButton | None = None
self.setWindowTitle("Grid Trading Manager")
self.resize(1280, 780)
@@ -63,16 +129,26 @@ class MainWindow(QMainWindow):
self.refresh_all()
def _build_ui(self) -> None:
self._apply_light_theme()
root = QWidget()
root.setObjectName("appRoot")
root_layout = QHBoxLayout(root)
root_layout.setContentsMargins(14, 14, 14, 14)
root_layout.setSpacing(14)
nav = QListWidget()
nav.setObjectName("sideNav")
nav.addItems(["账户总览", "持仓管理", "成交记录", "网格策略", "设置"])
nav.setFixedWidth(150)
nav.setCurrentRow(1)
root_layout.addWidget(nav)
content = QWidget()
content.setObjectName("contentPanel")
content_layout = QVBoxLayout(content)
content_layout.setContentsMargins(0, 0, 0, 0)
content_layout.setSpacing(12)
self.summary_labels = self._create_summary_cards()
content_layout.addLayout(self.summary_cards_layout)
content_layout.addLayout(self._create_toolbar())
@@ -80,64 +156,248 @@ class MainWindow(QMainWindow):
splitter = QSplitter(Qt.Orientation.Vertical)
self.holdings_table = QTableWidget(0, len(self.HOLDING_COLUMNS))
self.holdings_table.setHorizontalHeaderLabels(self.HOLDING_COLUMNS)
self.holdings_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.holdings_table.horizontalHeader().setStretchLastSection(True)
self.holdings_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self.holdings_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self._configure_table(self.holdings_table)
self.holdings_table.itemSelectionChanged.connect(self._on_holding_selected)
splitter.addWidget(self.holdings_table)
details = QSplitter(Qt.Orientation.Horizontal)
self.detail_box = QGroupBox("选中标的详情")
self.detail_layout = QGridLayout(self.detail_box)
self.detail_labels = {
"price_source": QLabel("-"),
"cost": QLabel("-"),
"breakeven": QLabel("-"),
"profit": QLabel("-"),
}
self.detail_layout.addWidget(QLabel("价格来源"), 0, 0)
self.detail_layout.addWidget(self.detail_labels["price_source"], 0, 1)
self.detail_layout.addWidget(QLabel("剩余成本"), 1, 0)
self.detail_layout.addWidget(self.detail_labels["cost"], 1, 1)
self.detail_layout.addWidget(QLabel("回本价"), 2, 0)
self.detail_layout.addWidget(self.detail_labels["breakeven"], 2, 1)
self.detail_layout.addWidget(QLabel("利润"), 3, 0)
self.detail_layout.addWidget(self.detail_labels["profit"], 3, 1)
self.details_tabs = QTabWidget()
self.details_tabs.setObjectName("detailsTabs")
trade_group = QGroupBox("最近成交")
trade_layout = QVBoxLayout(trade_group)
grid_tab = QWidget()
grid_layout = QVBoxLayout(grid_tab)
grid_layout.setContentsMargins(10, 10, 10, 10)
grid_layout.setSpacing(8)
grid_controls = QHBoxLayout()
self.grid_levels_count_edit = QSpinBox()
self.grid_levels_count_edit.setRange(1, 100)
self.grid_levels_count_edit.setValue(10)
self.grid_levels_count_edit.valueChanged.connect(lambda _value: self._refresh_details())
self.grid_levels_hint = QLabel("请先刷新行情")
grid_controls.addWidget(QLabel("档数"))
grid_controls.addWidget(self.grid_levels_count_edit)
grid_controls.addWidget(self.grid_levels_hint)
grid_controls.addStretch()
self.grid_levels_table = QTableWidget(0, len(self.GRID_LEVEL_COLUMNS))
self.grid_levels_table.setHorizontalHeaderLabels(self.GRID_LEVEL_COLUMNS)
self._configure_table(self.grid_levels_table)
grid_layout.addLayout(grid_controls)
grid_layout.addWidget(self.grid_levels_table)
open_grid_tab = QWidget()
open_grid_layout = QVBoxLayout(open_grid_tab)
open_grid_layout.setContentsMargins(10, 10, 10, 10)
open_grid_layout.setSpacing(8)
self.open_grid_lots_table = QTableWidget(0, len(self.OPEN_GRID_LOT_COLUMNS))
self.open_grid_lots_table.setHorizontalHeaderLabels(self.OPEN_GRID_LOT_COLUMNS)
self._configure_table(self.open_grid_lots_table)
open_grid_layout.addWidget(self.open_grid_lots_table)
grid_match_tab = QWidget()
grid_match_layout = QVBoxLayout(grid_match_tab)
grid_match_layout.setContentsMargins(10, 10, 10, 10)
grid_match_layout.setSpacing(8)
self.grid_matches_table = QTableWidget(0, len(self.GRID_MATCH_COLUMNS))
self.grid_matches_table.setHorizontalHeaderLabels(self.GRID_MATCH_COLUMNS)
self._configure_table(self.grid_matches_table)
grid_match_layout.addWidget(self.grid_matches_table)
trade_tab = QWidget()
trade_layout = QVBoxLayout(trade_tab)
trade_layout.setContentsMargins(10, 10, 10, 10)
trade_layout.setSpacing(8)
trade_buttons = QHBoxLayout()
edit_trade_button = QPushButton("编辑成交")
edit_trade_button.setObjectName("toolbarButton")
edit_trade_button.clicked.connect(self._edit_selected_trade)
delete_trade_button = QPushButton("删除成交")
delete_trade_button.setObjectName("toolbarButton")
delete_trade_button.clicked.connect(self._delete_selected_trade)
trade_buttons.addWidget(edit_trade_button)
trade_buttons.addWidget(delete_trade_button)
trade_buttons.addStretch()
self.trades_table = QTableWidget(0, len(self.TRADE_COLUMNS))
self.trades_table.setHorizontalHeaderLabels(self.TRADE_COLUMNS)
self.trades_table.horizontalHeader().setStretchLastSection(True)
self.trades_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self.trades_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self._configure_table(self.trades_table)
trade_layout.addLayout(trade_buttons)
trade_layout.addWidget(self.trades_table)
details.addWidget(self.detail_box)
details.addWidget(trade_group)
splitter.addWidget(details)
self.details_tabs.addTab(grid_tab, "网格档位")
self.details_tabs.addTab(open_grid_tab, "待卖网格")
self.details_tabs.addTab(grid_match_tab, "配对明细")
self.details_tabs.addTab(trade_tab, "最近成交")
self.details_tabs.setCurrentIndex(1)
splitter.addWidget(self.details_tabs)
splitter.setSizes([470, 230])
content_layout.addWidget(splitter)
root_layout.addWidget(content)
self.setCentralWidget(root)
def _apply_light_theme(self) -> None:
self.setStyleSheet(
"""
QMainWindow {
background: #f4f7fb;
color: #334155;
font-family: "Microsoft YaHei", "Segoe UI", Arial, sans-serif;
font-size: 13px;
}
QWidget#appRoot {
background: #f4f7fb;
}
QListWidget#sideNav {
background: #ffffff;
border: 1px solid #dbe4ef;
border-radius: 8px;
padding: 8px;
outline: 0;
}
QListWidget#sideNav::item {
border-radius: 6px;
color: #475569;
margin: 2px 0;
padding: 10px 12px;
}
QListWidget#sideNav::item:selected {
background: #e8f1ff;
color: #1d4ed8;
font-weight: 700;
}
QFrame#summaryCard {
background: #ffffff;
border: 1px solid #dbe4ef;
border-radius: 8px;
}
QLabel#summaryTitle {
color: #64748b;
font-size: 12px;
}
QLabel#summaryValue {
color: #0f172a;
font-size: 20px;
font-weight: 700;
}
QPushButton#toolbarButton {
background: #2563eb;
border: 1px solid #1d4ed8;
border-radius: 6px;
color: #ffffff;
font-weight: 700;
min-height: 30px;
padding: 6px 12px;
}
QPushButton#toolbarButton:hover {
background: #1d4ed8;
}
QPushButton#toolbarButton:pressed {
background: #1e40af;
}
QPushButton#toolbarButton:disabled {
background: #cbd5e1;
border-color: #cbd5e1;
color: #f8fafc;
}
QTabWidget#detailsTabs::pane {
background: #ffffff;
border: 1px solid #dbe4ef;
border-radius: 8px;
top: -1px;
}
QTabBar::tab {
background: #eef2f7;
border: 1px solid #dbe4ef;
border-bottom: 0;
border-top-left-radius: 6px;
border-top-right-radius: 6px;
color: #475569;
margin-right: 4px;
padding: 8px 14px;
}
QTabBar::tab:selected {
background: #ffffff;
color: #1d4ed8;
font-weight: 700;
}
QTableWidget {
background: #ffffff;
alternate-background-color: #f8fafc;
border: 1px solid #dbe4ef;
border-radius: 8px;
gridline-color: #e2e8f0;
selection-background-color: #dbeafe;
selection-color: #0f172a;
}
QHeaderView::section {
background: #f1f5f9;
border: 0;
border-right: 1px solid #dbe4ef;
border-bottom: 1px solid #dbe4ef;
color: #475569;
font-weight: 700;
padding: 8px 10px;
}
QSpinBox {
background: #ffffff;
border: 1px solid #cbd5e1;
border-radius: 6px;
min-height: 28px;
padding: 2px 8px;
}
QLabel {
color: #334155;
}
"""
)
def _configure_table(self, table: QTableWidget) -> None:
table.setAlternatingRowColors(True)
table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
table.setShowGrid(True)
table.setWordWrap(False)
table.verticalHeader().setVisible(False)
table.verticalHeader().setDefaultSectionSize(34)
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
table.horizontalHeader().setStretchLastSection(True)
def _make_item(self, value: str, pnl_value: Decimal | None = None) -> QTableWidgetItem:
item = QTableWidgetItem(value)
if pnl_value is not None:
self._style_pnl_item(item, pnl_value)
return item
def _style_pnl_item(self, item: QTableWidgetItem, value: Decimal | None) -> None:
if value is None:
return
font = QFont(item.font())
if value > 0:
item.setForeground(QBrush(self.PROFIT_COLOR))
font.setBold(True)
elif value < 0:
item.setForeground(QBrush(self.LOSS_COLOR))
font.setBold(True)
else:
font.setBold(False)
item.setFont(font)
def _style_pnl_label(self, label: QLabel, value: Decimal | None) -> None:
if value is None or value == 0:
label.setStyleSheet(f"color: {self.NEUTRAL_COLOR.name()}; font-weight: 400;")
elif value > 0:
label.setStyleSheet(f"color: {self.PROFIT_COLOR.name()}; font-weight: 700;")
else:
label.setStyleSheet(f"color: {self.LOSS_COLOR.name()}; font-weight: 700;")
def _create_summary_cards(self) -> dict[str, QLabel]:
self.summary_cards_layout = QGridLayout()
self.summary_cards_layout.setHorizontalSpacing(10)
self.summary_cards_layout.setVerticalSpacing(10)
labels: dict[str, QLabel] = {}
for column, (key, title) in enumerate(
[
("total_assets", "总资产"),
("total_assets", "账户权益"),
("cash", "现金"),
("market_value", "持仓市值"),
("floating_pnl", "浮动盈亏"),
@@ -145,11 +405,15 @@ class MainWindow(QMainWindow):
]
):
frame = QFrame()
frame.setObjectName("summaryCard")
frame.setFrameShape(QFrame.Shape.StyledPanel)
layout = QVBoxLayout(frame)
layout.setContentsMargins(14, 12, 14, 12)
layout.setSpacing(6)
title_label = QLabel(title)
title_label.setObjectName("summaryTitle")
value_label = QLabel("-")
value_label.setStyleSheet("font-size: 20px; font-weight: 700;")
value_label.setObjectName("summaryValue")
layout.addWidget(title_label)
layout.addWidget(value_label)
labels[key] = value_label
@@ -167,7 +431,10 @@ class MainWindow(QMainWindow):
]
for text, handler in buttons:
button = QPushButton(text)
button.setObjectName("toolbarButton")
button.clicked.connect(handler)
if text == "刷新行情":
self.refresh_quotes_button = button
layout.addWidget(button)
layout.addStretch()
return layout
@@ -183,18 +450,57 @@ class MainWindow(QMainWindow):
self.summary_labels["cash"].setText(format_money(account_summary.cash))
self.summary_labels["market_value"].setText(format_money(account_summary.market_value))
self.summary_labels["floating_pnl"].setText(format_money(account_summary.floating_pnl))
self._style_pnl_label(self.summary_labels["floating_pnl"], account_summary.floating_pnl)
self.summary_labels["usage"].setText(format_percent(account_summary.capital_usage_rate))
self._fill_holdings_table()
self._refresh_details()
def _refresh_quotes(self) -> None:
if self._quote_thread is not None and self._quote_thread.isRunning():
return
try:
self.service.refresh_quotes()
instruments = self.service.list_instruments()
except Exception as exc:
QMessageBox.warning(self, "行情刷新失败", str(exc))
return
if not instruments:
QMessageBox.information(self, "行情刷新", "请先添加股票或 ETF 标的。")
return
self._set_quote_refreshing(True)
self._quote_thread = QThread(self)
self._quote_worker = QuoteRefreshWorker(self.service, instruments)
self._quote_worker.moveToThread(self._quote_thread)
self._quote_thread.started.connect(self._quote_worker.run)
self._quote_worker.finished.connect(self._on_quote_refresh_finished)
self._quote_worker.failed.connect(self._on_quote_refresh_failed)
self._quote_worker.finished.connect(self._quote_thread.quit)
self._quote_worker.failed.connect(self._quote_thread.quit)
self._quote_thread.finished.connect(self._quote_worker.deleteLater)
self._quote_thread.finished.connect(self._on_quote_thread_finished)
self._quote_thread.start()
def _on_quote_refresh_finished(self, quotes: dict) -> None:
try:
self.service.apply_quote_snapshots(quotes)
except Exception as exc:
QMessageBox.warning(self, "行情刷新失败", str(exc))
return
self.refresh_all()
def _on_quote_refresh_failed(self, message: str) -> None:
QMessageBox.warning(self, "行情刷新失败", message)
def _on_quote_thread_finished(self) -> None:
self._set_quote_refreshing(False)
self._quote_thread = None
self._quote_worker = None
def _set_quote_refreshing(self, refreshing: bool) -> None:
if self.refresh_quotes_button is None:
return
self.refresh_quotes_button.setEnabled(not refreshing)
self.refresh_quotes_button.setText("刷新中..." if refreshing else "刷新行情")
def _fill_holdings_table(self) -> None:
self.holdings_table.setRowCount(len(self._positions))
for row, position in enumerate(self._positions):
@@ -213,8 +519,13 @@ class MainWindow(QMainWindow):
format_money(position.grid_profit),
format_money(position.floating_pnl),
]
pnl_values = {
10: position.realized_pnl,
11: position.grid_profit,
12: position.floating_pnl,
}
for column, value in enumerate(values):
item = QTableWidgetItem(value)
item = self._make_item(value, pnl_values.get(column))
item.setData(Qt.ItemDataRole.UserRole, position.instrument_id)
self.holdings_table.setItem(row, column, item)
if self._positions and self.holdings_table.currentRow() < 0:
@@ -229,20 +540,119 @@ class MainWindow(QMainWindow):
def _refresh_details(self) -> None:
position = self._selected_position()
if position is None:
for label in self.detail_labels.values():
label.setText("-")
self._refresh_grid_levels(None)
self._refresh_open_grid_lots(None)
self._refresh_grid_matches(None)
self._fill_trades_table([])
return
self.detail_labels["price_source"].setText(position.price_source)
self.detail_labels["cost"].setText(format_money(position.remaining_cost))
self.detail_labels["breakeven"].setText(
f"持仓 {format_price(position.position_breakeven_price)} / 账户 {format_price(position.account_breakeven_price)}"
)
self.detail_labels["profit"].setText(
f"已实现 {format_money(position.realized_pnl)} / 网格 {format_money(position.grid_profit)}"
)
self._refresh_grid_levels(position)
self._refresh_open_grid_lots(position)
self._refresh_grid_matches(position)
self._fill_trades_table(self.service.list_trades(instrument_id=position.instrument_id))
def _refresh_grid_levels(self, position: PositionSummary | None) -> None:
if position is None:
self.grid_levels_hint.setText("-")
self._fill_grid_levels_table([])
return
if position.current_price is None:
self.grid_levels_hint.setText("请先刷新行情")
self._fill_grid_levels_table([])
return
try:
levels = self.service.get_grid_level_suggestions(
position.instrument_id,
levels=self.grid_levels_count_edit.value(),
)
except Exception as exc:
QMessageBox.warning(self, "网格档位失败", str(exc))
self.grid_levels_hint.setText(str(exc))
self._fill_grid_levels_table([])
return
self.grid_levels_hint.setText("")
self._fill_grid_levels_table(levels)
def _fill_grid_levels_table(self, levels: list[GridLevelSuggestion]) -> None:
self.grid_levels_table.setRowCount(len(levels))
for row, level in enumerate(levels):
values = [
str(level.level),
format_price(level.buy_price),
format_money(level.buy_amount),
format_quantity(level.suggested_quantity),
format_money(level.actual_investment),
format_price(level.sell_price),
format_money(level.estimated_gross_profit),
]
for column, value in enumerate(values):
item = self._make_item(
value,
level.estimated_gross_profit if column == 6 else None,
)
self.grid_levels_table.setItem(row, column, item)
def _refresh_open_grid_lots(self, position: PositionSummary | None) -> None:
if position is None:
self._fill_open_grid_lots_table([])
return
try:
lots = self.service.get_open_grid_lots(position.instrument_id)
except Exception as exc:
QMessageBox.warning(self, "待卖网格失败", str(exc))
self._fill_open_grid_lots_table([])
return
self._fill_open_grid_lots_table(lots)
def _fill_open_grid_lots_table(self, lots: list[OpenGridLot]) -> None:
self.open_grid_lots_table.setRowCount(len(lots))
for row, lot in enumerate(lots):
values = [
lot.buy_date.isoformat(),
format_price(lot.buy_price),
format_quantity(lot.remaining_quantity),
format_money(lot.actual_investment),
format_price(lot.suggested_sell_price),
format_money(lot.estimated_gross_profit),
format_price(lot.current_price),
lot.status,
]
for column, value in enumerate(values):
item = self._make_item(
value,
lot.estimated_gross_profit if column == 5 else None,
)
self.open_grid_lots_table.setItem(row, column, item)
def _refresh_grid_matches(self, position: PositionSummary | None) -> None:
if position is None:
self._fill_grid_matches_table([])
return
try:
matches = self.service.get_grid_trade_matches(position.instrument_id)
except Exception as exc:
QMessageBox.warning(self, "配对明细失败", str(exc))
self._fill_grid_matches_table([])
return
self._fill_grid_matches_table(matches)
def _fill_grid_matches_table(self, matches: list[GridTradeMatch]) -> None:
self.grid_matches_table.setRowCount(len(matches))
for row, match in enumerate(matches):
values = [
match.sell_date.isoformat(),
format_price(match.sell_price),
match.buy_date.isoformat(),
format_price(match.buy_price),
format_quantity(match.matched_quantity),
format_money(match.buy_amount),
format_money(match.sell_amount),
format_money(match.gross_profit),
f"{match.buy_trade_id or '-'} -> 卖{match.sell_trade_id or '-'}",
]
for column, value in enumerate(values):
item = self._make_item(value, match.gross_profit if column == 7 else None)
self.grid_matches_table.setItem(row, column, item)
def _fill_trades_table(self, trades: list[Trade]) -> None:
self._trade_ids_by_row = {}
self.trades_table.setRowCount(len(trades))

View File

@@ -3,8 +3,13 @@ from decimal import Decimal
import pytest
from grid_trading.domain.calculations import CalculationError, calculate_positions, estimate_fees
from grid_trading.domain.models import FeeRules, Instrument, QuoteSnapshot, Trade, TradeGroup, TradeSide
from grid_trading.domain.calculations import (
CalculationError,
calculate_account_summary,
calculate_positions,
estimate_fees,
)
from grid_trading.domain.models import Account, FeeRules, Instrument, QuoteSnapshot, Trade, TradeGroup, TradeSide
def make_trade(
@@ -37,7 +42,7 @@ def make_trade(
def test_buy_sell_grid_profit_and_breakeven():
today = date(2026, 7, 8)
instrument = Instrument(id=1, code="510300", name="沪深300ETF", manual_price=Decimal("9.50"))
instrument = Instrument(id=1, code="510300", name="沪深300ETF")
trades = [
make_trade(
trade_id=1,
@@ -68,7 +73,17 @@ def test_buy_sell_grid_profit_and_breakeven():
),
]
[summary] = calculate_positions([instrument], trades, as_of=today)
quotes = {
1: QuoteSnapshot(
symbol="sh510300",
code="510300",
name="沪深300ETF",
price=Decimal("9.50"),
source="tencent",
)
}
[summary] = calculate_positions([instrument], trades, as_of=today, quote_snapshots=quotes)
assert summary.total_quantity == 100
assert summary.base_quantity == 100
@@ -79,6 +94,8 @@ def test_buy_sell_grid_profit_and_breakeven():
assert summary.cost_price == Decimal("9.01")
assert summary.position_breakeven_price == Decimal("8.03")
assert summary.account_breakeven_price == Decimal("8.03")
assert summary.current_price == Decimal("9.50")
assert summary.price_source == "tencent"
assert summary.market_value == Decimal("950.00")
assert summary.floating_pnl == Decimal("49.00")
@@ -111,7 +128,7 @@ def test_t_plus_one_available_quantity_excludes_today_buys():
assert summary.available_quantity == 200
def test_quote_snapshot_overrides_manual_price_for_position_value():
def test_quote_snapshot_supplies_current_price_for_position_value():
today = date(2026, 7, 8)
instrument = Instrument(id=1, code="510300", name="沪深300ETF", manual_price=Decimal("3.90"))
trades = [
@@ -141,6 +158,61 @@ def test_quote_snapshot_overrides_manual_price_for_position_value():
assert summary.market_value == Decimal("4120.00")
def test_missing_quote_does_not_use_manual_or_last_trade_price_for_current_price():
today = date(2026, 7, 8)
instrument = Instrument(id=1, code="510300", name="沪深300ETF", manual_price=Decimal("3.90"))
trades = [
make_trade(
trade_id=1,
trade_date=today - timedelta(days=1),
side=TradeSide.BUY,
price="4.00",
quantity=1000,
trade_group=TradeGroup.BASE,
)
]
[summary] = calculate_positions([instrument], trades, as_of=today)
assert summary.current_price is None
assert summary.price_source == "missing"
assert summary.market_value is None
assert summary.floating_pnl is None
def test_account_summary_caps_capital_usage_when_cash_is_negative():
today = date(2026, 7, 8)
account = Account(id=1, name="主账户", initial_cash=Decimal("100"))
instrument = Instrument(id=1, code="600588", name="用友网络")
trades = [
make_trade(
trade_id=1,
trade_date=today,
side=TradeSide.BUY,
price="1.50",
quantity=100,
trade_group=TradeGroup.BASE,
)
]
quotes = {
1: QuoteSnapshot(
symbol="sh600588",
code="600588",
name="用友网络",
price=Decimal("1.00"),
source="tencent",
)
}
positions = calculate_positions([instrument], trades, as_of=today, quote_snapshots=quotes)
summary = calculate_account_summary(account, positions, [], trades)
assert summary.cash == Decimal("-50.00")
assert summary.market_value == Decimal("100.00")
assert summary.total_assets == Decimal("50.00")
assert summary.capital_usage_rate == Decimal("1.0000")
def test_sell_more_than_group_position_raises():
today = date(2026, 7, 8)
instrument = Instrument(id=1, code="159915", name="创业板ETF")

69
tests/test_grid_levels.py Normal file
View File

@@ -0,0 +1,69 @@
from decimal import Decimal
import pytest
from grid_trading.domain.grid_levels import generate_grid_levels
def test_generate_grid_levels_compounds_percentage_spacing_and_rounds_values():
levels = generate_grid_levels(
current_price=Decimal("10.00"),
spacing=Decimal("0.03"),
amount_per_grid=Decimal("10000"),
lot_size=100,
levels=3,
)
assert [item.level for item in levels] == [1, 2, 3]
assert [item.buy_price for item in levels] == [Decimal("9.70"), Decimal("9.41"), Decimal("9.13")]
assert [item.buy_amount for item in levels] == [Decimal("10000.00")] * 3
assert [item.suggested_quantity for item in levels] == [1000, 1000, 1000]
assert [item.actual_investment for item in levels] == [
Decimal("9700.00"),
Decimal("9410.00"),
Decimal("9130.00"),
]
assert [item.sell_price for item in levels] == [Decimal("9.99"), Decimal("9.69"), Decimal("9.40")]
assert [item.estimated_gross_profit for item in levels] == [
Decimal("290.00"),
Decimal("280.00"),
Decimal("270.00"),
]
def test_generate_grid_levels_uses_zero_quantity_when_amount_cannot_buy_one_lot():
[level] = generate_grid_levels(
current_price=Decimal("10.00"),
spacing=Decimal("0.03"),
amount_per_grid=Decimal("500"),
lot_size=100,
levels=1,
)
assert level.buy_price == Decimal("9.70")
assert level.suggested_quantity == 0
assert level.actual_investment == Decimal("0.00")
assert level.estimated_gross_profit == Decimal("0.00")
@pytest.mark.parametrize(
("current_price", "spacing", "amount_per_grid", "lot_size", "levels"),
[
(Decimal("0"), Decimal("0.03"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("0"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("1"), Decimal("10000"), 100, 10),
(Decimal("10"), Decimal("0.03"), Decimal("0"), 100, 10),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 0, 10),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 100, 0),
(Decimal("10"), Decimal("0.03"), Decimal("10000"), 100, 101),
],
)
def test_generate_grid_levels_validates_inputs(current_price, spacing, amount_per_grid, lot_size, levels):
with pytest.raises(ValueError):
generate_grid_levels(
current_price=current_price,
spacing=spacing,
amount_per_grid=amount_per_grid,
lot_size=lot_size,
levels=levels,
)

View File

@@ -0,0 +1,151 @@
from datetime import date, timedelta
from decimal import Decimal
from grid_trading.domain.models import Trade, TradeGroup, TradeSide
from grid_trading.domain.open_grid_lots import calculate_grid_trade_matches, calculate_open_grid_lots
def make_trade(
*,
trade_id: int,
trade_date: date,
side: TradeSide,
price: str,
quantity: int,
trade_group: TradeGroup = TradeGroup.GRID,
) -> Trade:
return Trade(
id=trade_id,
account_id=1,
instrument_id=1,
trade_date=trade_date,
side=side,
price=Decimal(price),
quantity=quantity,
trade_group=trade_group,
)
def test_calculate_open_grid_lots_fifo_matches_grid_sells_against_grid_buys():
today = date(2026, 7, 9)
trades = [
make_trade(
trade_id=1,
trade_date=today - timedelta(days=3),
side=TradeSide.BUY,
price="10.00",
quantity=300,
),
make_trade(
trade_id=2,
trade_date=today - timedelta(days=2),
side=TradeSide.BUY,
price="9.50",
quantity=200,
),
make_trade(
trade_id=3,
trade_date=today - timedelta(days=1),
side=TradeSide.SELL,
price="10.30",
quantity=350,
),
make_trade(
trade_id=4,
trade_date=today,
side=TradeSide.BUY,
price="8.00",
quantity=100,
trade_group=TradeGroup.BASE,
),
]
lots = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=Decimal("9.80"),
as_of=today,
)
assert len(lots) == 1
[lot] = lots
assert lot.buy_trade_id == 2
assert lot.buy_date == today - timedelta(days=2)
assert lot.buy_price == Decimal("9.50")
assert lot.remaining_quantity == 150
assert lot.actual_investment == Decimal("1425.00")
assert lot.suggested_sell_price == Decimal("9.79")
assert lot.estimated_gross_profit == Decimal("43.50")
assert lot.current_price == Decimal("9.80")
assert lot.status == "可卖"
def test_calculate_grid_trade_matches_splits_sell_against_fifo_buy_lots():
today = date(2026, 7, 9)
trades = [
make_trade(
trade_id=1,
trade_date=today - timedelta(days=3),
side=TradeSide.BUY,
price="10.00",
quantity=300,
),
make_trade(
trade_id=2,
trade_date=today - timedelta(days=2),
side=TradeSide.BUY,
price="9.50",
quantity=200,
),
make_trade(
trade_id=3,
trade_date=today - timedelta(days=1),
side=TradeSide.SELL,
price="10.30",
quantity=350,
),
]
matches = calculate_grid_trade_matches(trades, as_of=today)
assert len(matches) == 2
assert matches[0].sell_trade_id == 3
assert matches[0].sell_date == today - timedelta(days=1)
assert matches[0].sell_price == Decimal("10.30")
assert matches[0].buy_trade_id == 1
assert matches[0].buy_date == today - timedelta(days=3)
assert matches[0].buy_price == Decimal("10.00")
assert matches[0].matched_quantity == 300
assert matches[0].buy_amount == Decimal("3000.00")
assert matches[0].sell_amount == Decimal("3090.00")
assert matches[0].gross_profit == Decimal("90.00")
assert matches[1].buy_trade_id == 2
assert matches[1].matched_quantity == 50
assert matches[1].buy_amount == Decimal("475.00")
assert matches[1].sell_amount == Decimal("515.00")
assert matches[1].gross_profit == Decimal("40.00")
def test_calculate_open_grid_lots_reports_not_reached_and_missing_quote_statuses():
today = date(2026, 7, 9)
trades = [
make_trade(trade_id=1, trade_date=today, side=TradeSide.BUY, price="10.00", quantity=100),
]
[not_reached] = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=Decimal("10.20"),
as_of=today,
)
[missing_quote] = calculate_open_grid_lots(
trades,
spacing=Decimal("0.03"),
current_price=None,
as_of=today,
)
assert not_reached.suggested_sell_price == Decimal("10.30")
assert not_reached.status == "未到价"
assert missing_quote.current_price is None
assert missing_quote.status == "未刷新行情"

View File

@@ -64,10 +64,12 @@ def test_service_creates_default_account_and_computes_summary(tmp_path):
assert len(positions) == 1
assert positions[0].total_quantity == 500
assert positions[0].current_price is None
assert positions[0].price_source == "missing"
assert positions[0].grid_profit == Decimal("91.47")
assert summary.cash == Decimal("98138.97")
assert summary.market_value == Decimal("2000.00")
assert summary.total_assets == Decimal("100138.97")
assert summary.market_value == Decimal("0.00")
assert summary.total_assets == Decimal("98138.97")
def test_service_validates_lot_size_and_available_sell_quantity(tmp_path):
@@ -221,3 +223,130 @@ def test_service_refresh_quotes_uses_realtime_price_in_summaries(tmp_path):
assert position.current_price == Decimal("4.12")
assert position.price_source == "tencent"
assert summary.market_value == Decimal("4120.00")
def test_service_generates_grid_level_suggestions_from_realtime_price(tmp_path):
service = TradingService(tmp_path / "grid.db", quote_provider=FakeQuoteProvider())
service.ensure_defaults()
account = service.get_active_account()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("4.00"),
quantity=1000,
trade_group=TradeGroup.BASE,
)
)
service.refresh_quotes()
levels = service.get_grid_level_suggestions(instrument.id, levels=2)
assert [item.buy_price for item in levels] == [Decimal("4.00"), Decimal("3.88")]
assert [item.sell_price for item in levels] == [Decimal("4.12"), Decimal("4.00")]
assert [item.buy_amount for item in levels] == [Decimal("5000.00"), Decimal("5000.00")]
assert [item.suggested_quantity for item in levels] == [1200, 1200]
def test_service_returns_empty_grid_levels_without_realtime_price(tmp_path):
service = TradingService(tmp_path / "grid.db")
service.ensure_defaults()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
assert service.get_grid_level_suggestions(instrument.id, levels=10) == []
def test_service_returns_open_grid_lots_with_suggested_sell_price(tmp_path):
service = TradingService(tmp_path / "grid.db", quote_provider=FakeQuoteProvider())
service.ensure_defaults()
account = service.get_active_account()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("4.00"),
quantity=1000,
trade_group=TradeGroup.GRID,
)
)
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 8),
side=TradeSide.SELL,
price=Decimal("4.12"),
quantity=400,
trade_group=TradeGroup.GRID,
)
)
service.refresh_quotes()
[lot] = service.get_open_grid_lots(instrument.id, as_of=date(2026, 7, 9))
assert lot.buy_price == Decimal("4.00")
assert lot.remaining_quantity == 600
assert lot.suggested_sell_price == Decimal("4.12")
assert lot.current_price == Decimal("4.12")
assert lot.status == "可卖"
def test_service_returns_grid_trade_matches(tmp_path):
service = TradingService(tmp_path / "grid.db")
service.ensure_defaults()
account = service.get_active_account()
instrument = service.add_instrument(Instrument(id=None, code="510300", name="沪深300ETF", market="ETF"))
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 6),
side=TradeSide.BUY,
price=Decimal("10.00"),
quantity=300,
trade_group=TradeGroup.GRID,
)
)
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("9.50"),
quantity=200,
trade_group=TradeGroup.GRID,
)
)
service.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 8),
side=TradeSide.SELL,
price=Decimal("10.30"),
quantity=400,
trade_group=TradeGroup.GRID,
)
)
matches = service.get_grid_trade_matches(instrument.id, as_of=date(2026, 7, 9))
assert len(matches) == 2
assert matches[0].buy_trade_id is not None
assert matches[0].matched_quantity == 300
assert matches[0].gross_profit == Decimal("90.00")
assert matches[1].matched_quantity == 100
assert matches[1].gross_profit == Decimal("80.00")

View File

@@ -74,3 +74,25 @@ def test_provider_wraps_fetch_errors():
with pytest.raises(QuoteFetchError, match="Tencent quote request failed"):
provider.fetch_quotes([Instrument(id=1, code="000001", name="平安银行")])
def test_provider_retries_once_after_transient_fetch_error():
raw = (
'v_sz000001="51~平安银行~000001~10.60~10.47~10.44~0~0~0~'
'0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~~20260708161430~0.13~1.24~10.63~10.34";'
).encode("gbk")
calls = 0
def flaky_fetch(url: str, timeout: float) -> bytes:
nonlocal calls
calls += 1
if calls == 1:
raise OSError("bad gateway")
return raw
provider = TencentQuoteProvider(fetcher=flaky_fetch, retries=1)
quotes = provider.fetch_quotes([Instrument(id=1, code="000001", name="平安银行")])
assert calls == 2
assert quotes["000001"].price == Decimal("10.60")

View File

@@ -1,4 +1,7 @@
from decimal import Decimal
import time
from grid_trading.domain.models import Instrument
def test_formatters_render_money_percent_and_empty_values():
@@ -14,7 +17,7 @@ def test_formatters_render_money_percent_and_empty_values():
def test_main_window_can_be_constructed_offscreen(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QPushButton
from PySide6.QtWidgets import QApplication, QLabel, QPushButton
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
@@ -26,6 +29,228 @@ def test_main_window_can_be_constructed_offscreen(tmp_path, monkeypatch):
assert window.windowTitle() == "Grid Trading Manager"
assert window.holdings_table.columnCount() > 0
assert any(button.text() == "刷新行情" for button in window.findChildren(QPushButton))
summary_titles = [label.text() for label in window.findChildren(QLabel)]
assert "账户权益" in summary_titles
assert "总资产" not in summary_titles
window.close()
service.close()
app.processEvents()
def test_main_window_contains_grid_level_table(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QTabWidget
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.grid_levels_table.columnCount() == 7
tab_titles = _tab_titles(window.details_tabs)
assert tab_titles == ["网格档位", "待卖网格", "配对明细", "最近成交"]
assert isinstance(window.details_tabs, QTabWidget)
window.close()
service.close()
app.processEvents()
def test_main_window_contains_open_grid_lots_table(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.open_grid_lots_table.columnCount() == 8
assert "待卖网格" in _tab_titles(window.details_tabs)
assert window.grid_matches_table.columnCount() == 9
assert "配对明细" in _tab_titles(window.details_tabs)
window.close()
service.close()
app.processEvents()
def test_main_window_applies_light_theme(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert window.centralWidget().objectName() == "appRoot"
assert window.holdings_table.alternatingRowColors()
assert "QMainWindow" in window.styleSheet()
window.close()
service.close()
app.processEvents()
def test_profit_and_loss_cells_use_a_share_colors(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtGui import QColor
from PySide6.QtWidgets import QApplication
from grid_trading.domain.models import PositionSummary
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
window._positions = [
PositionSummary(
instrument_id=1,
code="510300",
name="沪深300ETF",
market="ETF",
current_price=Decimal("10.50"),
price_source="quote",
total_quantity=100,
available_quantity=100,
base_quantity=0,
grid_quantity=100,
other_quantity=0,
remaining_cost=Decimal("1000"),
cost_price=Decimal("10"),
position_breakeven_price=Decimal("10"),
account_breakeven_price=Decimal("10"),
realized_pnl=Decimal("12.34"),
grid_profit=Decimal("-5.67"),
floating_pnl=Decimal("50"),
market_value=Decimal("1050"),
)
]
window.holdings_table.blockSignals(True)
window._fill_holdings_table()
window.holdings_table.blockSignals(False)
profit_item = window.holdings_table.item(0, 10)
loss_item = window.holdings_table.item(0, 11)
assert profit_item.foreground().color() == QColor("#c62828")
assert profit_item.font().bold()
assert loss_item.foreground().color() == QColor("#2e7d32")
assert loss_item.font().bold()
window.close()
service.close()
app.processEvents()
def test_main_window_removes_selected_instrument_detail_group(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QGroupBox
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
window = MainWindow(service)
assert not any(group.title() == "选中标的详情" for group in window.findChildren(QGroupBox))
window.close()
service.close()
app.processEvents()
def test_instrument_dialog_does_not_collect_manual_current_price(monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QLabel
from grid_trading.ui.dialogs import InstrumentDialog
app = QApplication.instance() or QApplication([])
dialog = InstrumentDialog()
labels = [label.text() for label in dialog.findChildren(QLabel)]
assert "手动价格" not in labels
assert not hasattr(dialog, "manual_price_edit")
assert dialog.to_instrument().manual_price is None
dialog.close()
app.processEvents()
def test_trade_dialog_labels_price_as_trade_price(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication, QLabel
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.dialogs import TradeDialog
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db")
service.ensure_defaults()
account = service.get_active_account()
dialog = TradeDialog(service, account.id, [Instrument(id=1, code="510300", name="沪深300ETF")])
labels = [label.text() for label in dialog.findChildren(QLabel)]
assert "成交价" in labels
dialog.close()
service.close()
app.processEvents()
def _tab_titles(tabs):
return [tabs.tabText(index) for index in range(tabs.count())]
def test_quote_refresh_does_not_block_main_window(tmp_path, monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from grid_trading.services.trading_service import TradingService
from grid_trading.ui.main_window import MainWindow
class SlowQuoteProvider:
def fetch_quotes(self, instruments):
time.sleep(0.3)
return {}
app = QApplication.instance() or QApplication([])
service = TradingService(tmp_path / "grid.db", quote_provider=SlowQuoteProvider())
service.ensure_defaults()
service.add_instrument(Instrument(id=None, code="000001", name="Ping An Bank"))
window = MainWindow(service)
started_at = time.perf_counter()
window._refresh_quotes()
elapsed = time.perf_counter() - started_at
assert elapsed < 0.15
deadline = time.perf_counter() + 2
while getattr(window, "_quote_thread", None) is not None and time.perf_counter() < deadline:
app.processEvents()
time.sleep(0.01)
assert getattr(window, "_quote_thread", None) is None
window.close()
service.close()