Grid_Trading/docs/superpowers/plans/2026-07-09-open-grid-lots.md

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:

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:

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:

@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:

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:

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:

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:

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:

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:

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:

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:

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:

OPEN_GRID_LOT_COLUMNS = [
    "买入日期",
    "买入价",
    "剩余股数",
    "实际投入",
    "建议卖出价",
    "预计毛利润",
    "当前价",
    "状态",
]

Create self.open_grid_lots_table in _build_ui, add a QGroupBox("待卖网格"), and refresh it from _refresh_details:

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:

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:

pytest -q

Expected: all tests pass.

• Step 3: Run CLI smoke

Run:

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.