wangpeng/Grid_Trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: show open grid lots

Browse Source
This commit is contained in:
王鹏
2026-07-09 11:17:51 +08:00
parent fe11929aa0
commit 5d66423895
8 changed files with 338 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@@ -34,6 +34,10 @@ http://qt.gtimg.cn/q=<symbol>
- 建议买入股数按标的交易单位向下取整。
- 预计单轮毛利润不扣手续费，生成结果只用于展示，不会写入成交记录或 SQLite。
## 待卖网格说明
底部“待卖网格”表会展示尚未被网格卖出抵消的网格买入批次，包括买入价、剩余股数、建议卖出价、预计毛利润、当前价和状态。网格卖出按时间顺序 FIFO 抵消最早的网格买入；建议卖出价按 `买入价 * (1 + 网格间距)` 计算。
## 开发环境
```powershell

13
src/grid_trading/domain/models.py
View File

@@ -125,6 +125,19 @@ class GridLevelSuggestion:
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 QuoteSnapshot:
symbol: str

77
src/grid_trading/domain/open_grid_lots.py Normal file
View File

@@ -0,0 +1,77 @@
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:
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 _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 "未到价"

28
src/grid_trading/services/trading_service.py
View File

@@ -20,6 +20,7 @@ from grid_trading.domain.models import (
FeeRules,
GridLevelSuggestion,
Instrument,
OpenGridLot,
PositionSummary,
QuoteSnapshot,
StrategyOverride,
@@ -27,6 +28,7 @@ from grid_trading.domain.models import (
Trade,
TradeSide,
)
from grid_trading.domain.open_grid_lots import calculate_open_grid_lots
from grid_trading.market.tencent import TencentQuoteProvider
from grid_trading.storage.repositories import Repository
@@ -234,6 +236,32 @@ class TradingService:
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 _require_account(self, account_id: int) -> Account:
account = self.repository.get_account(account_id)
if account is None:

52
src/grid_trading/ui/main_window.py
View File

@@ -26,7 +26,7 @@ from PySide6.QtWidgets import (
)
from grid_trading.config import DEFAULT_DB_PATH
from grid_trading.domain.models import GridLevelSuggestion, PositionSummary, Trade, TradeSide
from grid_trading.domain.models import GridLevelSuggestion, 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
@@ -76,6 +76,16 @@ class MainWindow(QMainWindow):
"卖出价",
"预计单轮毛利润",
]
OPEN_GRID_LOT_COLUMNS = [
"买入日期",
"买入价",
"剩余股数",
"实际投入",
"建议卖出价",
"预计毛利润",
"当前价",
"状态",
]
def __init__(self, service: TradingService):
super().__init__()
@@ -156,6 +166,15 @@ class MainWindow(QMainWindow):
grid_layout.addLayout(grid_controls)
grid_layout.addWidget(self.grid_levels_table)
open_grid_group = QGroupBox("待卖网格")
open_grid_layout = QVBoxLayout(open_grid_group)
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.open_grid_lots_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.open_grid_lots_table.horizontalHeader().setStretchLastSection(True)
self.open_grid_lots_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
open_grid_layout.addWidget(self.open_grid_lots_table)
trade_group = QGroupBox("最近成交")
trade_layout = QVBoxLayout(trade_group)
trade_buttons = QHBoxLayout()
@@ -176,6 +195,7 @@ class MainWindow(QMainWindow):
details.addWidget(self.detail_box)
details.addWidget(grid_group)
details.addWidget(open_grid_group)
details.addWidget(trade_group)
splitter.addWidget(details)
splitter.setSizes([470, 230])
@@ -324,6 +344,7 @@ class MainWindow(QMainWindow):
for label in self.detail_labels.values():
label.setText("-")
self._refresh_grid_levels(None)
self._refresh_open_grid_lots(None)
self._fill_trades_table([])
return
self.detail_labels["price_source"].setText(position.price_source)
@@ -335,6 +356,7 @@ class MainWindow(QMainWindow):
f"已实现 {format_money(position.realized_pnl)} / 网格 {format_money(position.grid_profit)}"
)
self._refresh_grid_levels(position)
self._refresh_open_grid_lots(position)
self._fill_trades_table(self.service.list_trades(instrument_id=position.instrument_id))
def _refresh_grid_levels(self, position: PositionSummary | None) -> None:
@@ -374,6 +396,34 @@ class MainWindow(QMainWindow):
for column, value in enumerate(values):
self.grid_levels_table.setItem(row, column, QTableWidgetItem(value))
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):
self.open_grid_lots_table.setItem(row, column, QTableWidgetItem(value))
def _fill_trades_table(self, trades: list[Trade]) -> None:
self._trade_ids_by_row = {}
self.trades_table.setRowCount(len(trades))

105
tests/test_open_grid_lots.py Normal file
View File

@@ -0,0 +1,105 @@
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: 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_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 == "未刷新行情"

40
tests/test_services.py
View File

@@ -258,3 +258,43 @@ def test_service_returns_empty_grid_levels_without_realtime_price(tmp_path):
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 == "可卖"

20
tests/test_ui.py
View File

@@ -55,6 +55,26 @@ def test_main_window_contains_grid_level_table(tmp_path, monkeypatch):
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, 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()
def test_instrument_dialog_does_not_collect_manual_current_price(monkeypatch):
monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen")