feat: add sqlite persistence

This commit is contained in:
王鹏
2026-07-08 17:23:24 +08:00
parent 16c0d5a53a
commit 0f4dfead6f
4 changed files with 654 additions and 0 deletions

View File

@@ -0,0 +1 @@
"""SQLite persistence layer."""

View File

@@ -0,0 +1,104 @@
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA = """
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
initial_cash TEXT NOT NULL,
notes TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS instruments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT NOT NULL,
name TEXT NOT NULL,
market TEXT NOT NULL DEFAULT 'A',
lot_size INTEGER NOT NULL DEFAULT 100,
manual_price TEXT,
allow_odd_lot INTEGER NOT NULL DEFAULT 0,
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(code, market)
);
CREATE TABLE IF NOT EXISTS strategy_templates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
grid_spacing_pct TEXT NOT NULL,
amount_per_grid TEXT NOT NULL,
base_target_amount TEXT NOT NULL,
max_position_amount TEXT NOT NULL,
min_lot INTEGER NOT NULL DEFAULT 100,
is_default INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS instrument_strategy_overrides (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument_id INTEGER NOT NULL UNIQUE REFERENCES instruments(id) ON DELETE CASCADE,
template_id INTEGER REFERENCES strategy_templates(id) ON DELETE SET NULL,
grid_spacing_pct TEXT,
amount_per_grid TEXT,
base_target_amount TEXT,
max_position_amount TEXT,
min_lot INTEGER,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
instrument_id INTEGER NOT NULL REFERENCES instruments(id) ON DELETE CASCADE,
trade_date TEXT NOT NULL,
side TEXT NOT NULL,
price TEXT NOT NULL,
quantity INTEGER NOT NULL,
commission TEXT NOT NULL DEFAULT '0',
stamp_tax TEXT NOT NULL DEFAULT '0',
transfer_fee TEXT NOT NULL DEFAULT '0',
trade_group TEXT NOT NULL,
notes TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS cash_ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
instrument_id INTEGER REFERENCES instruments(id) ON DELETE SET NULL,
entry_date TEXT NOT NULL,
amount TEXT NOT NULL,
category TEXT NOT NULL,
notes TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"""
def connect_database(db_path: str | Path) -> sqlite3.Connection:
path = Path(db_path)
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
return connection
def initialize_database(connection: sqlite3.Connection) -> None:
connection.executescript(SCHEMA)
connection.commit()

View File

@@ -0,0 +1,444 @@
from __future__ import annotations
import sqlite3
from datetime import date
from decimal import Decimal
from pathlib import Path
from typing import Sequence
from grid_trading.domain.models import (
Account,
CashLedgerEntry,
Instrument,
StrategyOverride,
StrategyTemplate,
Trade,
TradeGroup,
TradeSide,
)
from grid_trading.storage.database import connect_database, initialize_database
class Repository:
def __init__(self, db_path: str | Path):
self.db_path = Path(db_path)
self.connection = connect_database(self.db_path)
def initialize(self) -> None:
initialize_database(self.connection)
def close(self) -> None:
self.connection.close()
def save_account(self, account: Account) -> Account:
if account.id is None:
cursor = self.connection.execute(
"INSERT INTO accounts (name, initial_cash, notes) VALUES (?, ?, ?)",
(account.name, _decimal_to_text(account.initial_cash), account.notes),
)
self.connection.commit()
return Account(
id=cursor.lastrowid,
name=account.name,
initial_cash=account.initial_cash,
notes=account.notes,
)
self.connection.execute(
"""
UPDATE accounts
SET name = ?, initial_cash = ?, notes = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(account.name, _decimal_to_text(account.initial_cash), account.notes, account.id),
)
self.connection.commit()
return account
def list_accounts(self) -> list[Account]:
rows = self.connection.execute("SELECT * FROM accounts ORDER BY id").fetchall()
return [_row_to_account(row) for row in rows]
def get_account(self, account_id: int) -> Account | None:
row = self.connection.execute("SELECT * FROM accounts WHERE id = ?", (account_id,)).fetchone()
return _row_to_account(row) if row else None
def save_instrument(self, instrument: Instrument) -> Instrument:
values = (
instrument.code,
instrument.name,
instrument.market,
instrument.lot_size,
_optional_decimal_to_text(instrument.manual_price),
int(instrument.allow_odd_lot),
int(instrument.active),
)
if instrument.id is None:
cursor = self.connection.execute(
"""
INSERT INTO instruments
(code, name, market, lot_size, manual_price, allow_odd_lot, active)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
values,
)
self.connection.commit()
return Instrument(
id=cursor.lastrowid,
code=instrument.code,
name=instrument.name,
market=instrument.market,
lot_size=instrument.lot_size,
manual_price=instrument.manual_price,
allow_odd_lot=instrument.allow_odd_lot,
active=instrument.active,
)
self.connection.execute(
"""
UPDATE instruments
SET code = ?, name = ?, market = ?, lot_size = ?, manual_price = ?,
allow_odd_lot = ?, active = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(*values, instrument.id),
)
self.connection.commit()
return instrument
def list_instruments(self, *, include_inactive: bool = False) -> list[Instrument]:
if include_inactive:
rows = self.connection.execute("SELECT * FROM instruments ORDER BY code").fetchall()
else:
rows = self.connection.execute(
"SELECT * FROM instruments WHERE active = 1 ORDER BY code"
).fetchall()
return [_row_to_instrument(row) for row in rows]
def get_instrument(self, instrument_id: int) -> Instrument | None:
row = self.connection.execute(
"SELECT * FROM instruments WHERE id = ?",
(instrument_id,),
).fetchone()
return _row_to_instrument(row) if row else None
def get_instrument_by_code(self, code: str, market: str = "A") -> Instrument | None:
row = self.connection.execute(
"SELECT * FROM instruments WHERE code = ? AND market = ?",
(code, market),
).fetchone()
return _row_to_instrument(row) if row else None
def save_strategy_template(self, template: StrategyTemplate) -> StrategyTemplate:
if template.is_default:
self.connection.execute("UPDATE strategy_templates SET is_default = 0")
values = (
template.name,
_decimal_to_text(template.grid_spacing_pct),
_decimal_to_text(template.amount_per_grid),
_decimal_to_text(template.base_target_amount),
_decimal_to_text(template.max_position_amount),
template.min_lot,
int(template.is_default),
)
if template.id is None:
cursor = self.connection.execute(
"""
INSERT INTO strategy_templates
(name, grid_spacing_pct, amount_per_grid, base_target_amount,
max_position_amount, min_lot, is_default)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
values,
)
self.connection.commit()
return StrategyTemplate(
id=cursor.lastrowid,
name=template.name,
grid_spacing_pct=template.grid_spacing_pct,
amount_per_grid=template.amount_per_grid,
base_target_amount=template.base_target_amount,
max_position_amount=template.max_position_amount,
min_lot=template.min_lot,
is_default=template.is_default,
)
self.connection.execute(
"""
UPDATE strategy_templates
SET name = ?, grid_spacing_pct = ?, amount_per_grid = ?,
base_target_amount = ?, max_position_amount = ?, min_lot = ?,
is_default = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(*values, template.id),
)
self.connection.commit()
return template
def list_strategy_templates(self) -> list[StrategyTemplate]:
rows = self.connection.execute("SELECT * FROM strategy_templates ORDER BY id").fetchall()
return [_row_to_strategy_template(row) for row in rows]
def get_default_strategy_template(self) -> StrategyTemplate | None:
row = self.connection.execute(
"SELECT * FROM strategy_templates WHERE is_default = 1 ORDER BY id LIMIT 1"
).fetchone()
return _row_to_strategy_template(row) if row else None
def save_strategy_override(self, override: StrategyOverride) -> StrategyOverride:
self.connection.execute(
"""
INSERT INTO instrument_strategy_overrides
(instrument_id, template_id, grid_spacing_pct, amount_per_grid,
base_target_amount, max_position_amount, min_lot)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(instrument_id) DO UPDATE SET
template_id = excluded.template_id,
grid_spacing_pct = excluded.grid_spacing_pct,
amount_per_grid = excluded.amount_per_grid,
base_target_amount = excluded.base_target_amount,
max_position_amount = excluded.max_position_amount,
min_lot = excluded.min_lot,
updated_at = CURRENT_TIMESTAMP
""",
(
override.instrument_id,
override.template_id,
_optional_decimal_to_text(override.grid_spacing_pct),
_optional_decimal_to_text(override.amount_per_grid),
_optional_decimal_to_text(override.base_target_amount),
_optional_decimal_to_text(override.max_position_amount),
override.min_lot,
),
)
self.connection.commit()
return override
def get_strategy_override(self, instrument_id: int) -> StrategyOverride | None:
row = self.connection.execute(
"SELECT * FROM instrument_strategy_overrides WHERE instrument_id = ?",
(instrument_id,),
).fetchone()
return _row_to_strategy_override(row) if row else None
def save_trade(self, trade: Trade) -> Trade:
values = (
trade.account_id,
trade.instrument_id,
trade.trade_date.isoformat(),
trade.side.value,
_decimal_to_text(trade.price),
trade.quantity,
_decimal_to_text(trade.commission),
_decimal_to_text(trade.stamp_tax),
_decimal_to_text(trade.transfer_fee),
trade.trade_group.value,
trade.notes,
)
if trade.id is None:
cursor = self.connection.execute(
"""
INSERT INTO trades
(account_id, instrument_id, trade_date, side, price, quantity,
commission, stamp_tax, transfer_fee, trade_group, notes)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
values,
)
self.connection.commit()
return Trade(
id=cursor.lastrowid,
account_id=trade.account_id,
instrument_id=trade.instrument_id,
trade_date=trade.trade_date,
side=trade.side,
price=trade.price,
quantity=trade.quantity,
commission=trade.commission,
stamp_tax=trade.stamp_tax,
transfer_fee=trade.transfer_fee,
trade_group=trade.trade_group,
notes=trade.notes,
)
self.connection.execute(
"""
UPDATE trades
SET account_id = ?, instrument_id = ?, trade_date = ?, side = ?, price = ?,
quantity = ?, commission = ?, stamp_tax = ?, transfer_fee = ?,
trade_group = ?, notes = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(*values, trade.id),
)
self.connection.commit()
return trade
def get_trade(self, trade_id: int | None) -> Trade | None:
if trade_id is None:
return None
row = self.connection.execute("SELECT * FROM trades WHERE id = ?", (trade_id,)).fetchone()
return _row_to_trade(row) if row else None
def list_trades(
self,
*,
account_id: int | None = None,
instrument_id: int | None = None,
) -> list[Trade]:
clauses: list[str] = []
params: list[int] = []
if account_id is not None:
clauses.append("account_id = ?")
params.append(account_id)
if instrument_id is not None:
clauses.append("instrument_id = ?")
params.append(instrument_id)
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
rows = self.connection.execute(
f"SELECT * FROM trades {where} ORDER BY trade_date, id",
params,
).fetchall()
return [_row_to_trade(row) for row in rows]
def delete_trade(self, trade_id: int) -> None:
self.connection.execute("DELETE FROM trades WHERE id = ?", (trade_id,))
self.connection.commit()
def save_cash_ledger_entry(self, entry: CashLedgerEntry) -> CashLedgerEntry:
if entry.id is not None:
raise ValueError("Cash ledger entries are append-only")
cursor = self.connection.execute(
"""
INSERT INTO cash_ledger
(account_id, instrument_id, entry_date, amount, category, notes)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
entry.account_id,
entry.instrument_id,
entry.entry_date.isoformat(),
_decimal_to_text(entry.amount),
entry.category,
entry.notes,
),
)
self.connection.commit()
return CashLedgerEntry(
id=cursor.lastrowid,
account_id=entry.account_id,
instrument_id=entry.instrument_id,
entry_date=entry.entry_date,
amount=entry.amount,
category=entry.category,
notes=entry.notes,
)
def list_cash_ledger_entries(self, *, account_id: int | None = None) -> list[CashLedgerEntry]:
if account_id is None:
rows = self.connection.execute("SELECT * FROM cash_ledger ORDER BY entry_date, id").fetchall()
else:
rows = self.connection.execute(
"SELECT * FROM cash_ledger WHERE account_id = ? ORDER BY entry_date, id",
(account_id,),
).fetchall()
return [_row_to_cash_entry(row) for row in rows]
def set_setting(self, key: str, value: str) -> None:
self.connection.execute(
"""
INSERT INTO app_settings (key, value) VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""",
(key, value),
)
self.connection.commit()
def get_setting(self, key: str, default: str | None = None) -> str | None:
row = self.connection.execute("SELECT value FROM app_settings WHERE key = ?", (key,)).fetchone()
return row["value"] if row else default
def _decimal_to_text(value: Decimal) -> str:
return str(value)
def _optional_decimal_to_text(value: Decimal | None) -> str | None:
return str(value) if value is not None else None
def _text_to_decimal(value: str | None) -> Decimal | None:
return Decimal(value) if value is not None else None
def _row_to_account(row: sqlite3.Row) -> Account:
return Account(
id=row["id"],
name=row["name"],
initial_cash=Decimal(row["initial_cash"]),
notes=row["notes"],
)
def _row_to_instrument(row: sqlite3.Row) -> Instrument:
return Instrument(
id=row["id"],
code=row["code"],
name=row["name"],
market=row["market"],
lot_size=row["lot_size"],
manual_price=_text_to_decimal(row["manual_price"]),
allow_odd_lot=bool(row["allow_odd_lot"]),
active=bool(row["active"]),
)
def _row_to_strategy_template(row: sqlite3.Row) -> StrategyTemplate:
return StrategyTemplate(
id=row["id"],
name=row["name"],
grid_spacing_pct=Decimal(row["grid_spacing_pct"]),
amount_per_grid=Decimal(row["amount_per_grid"]),
base_target_amount=Decimal(row["base_target_amount"]),
max_position_amount=Decimal(row["max_position_amount"]),
min_lot=row["min_lot"],
is_default=bool(row["is_default"]),
)
def _row_to_strategy_override(row: sqlite3.Row) -> StrategyOverride:
return StrategyOverride(
instrument_id=row["instrument_id"],
template_id=row["template_id"],
grid_spacing_pct=_text_to_decimal(row["grid_spacing_pct"]),
amount_per_grid=_text_to_decimal(row["amount_per_grid"]),
base_target_amount=_text_to_decimal(row["base_target_amount"]),
max_position_amount=_text_to_decimal(row["max_position_amount"]),
min_lot=row["min_lot"],
)
def _row_to_trade(row: sqlite3.Row) -> Trade:
return Trade(
id=row["id"],
account_id=row["account_id"],
instrument_id=row["instrument_id"],
trade_date=date.fromisoformat(row["trade_date"]),
side=TradeSide(row["side"]),
price=Decimal(row["price"]),
quantity=row["quantity"],
commission=Decimal(row["commission"]),
stamp_tax=Decimal(row["stamp_tax"]),
transfer_fee=Decimal(row["transfer_fee"]),
trade_group=TradeGroup(row["trade_group"]),
notes=row["notes"],
)
def _row_to_cash_entry(row: sqlite3.Row) -> CashLedgerEntry:
return CashLedgerEntry(
id=row["id"],
account_id=row["account_id"],
instrument_id=row["instrument_id"],
entry_date=date.fromisoformat(row["entry_date"]),
amount=Decimal(row["amount"]),
category=row["category"],
notes=row["notes"],
)

105
tests/test_repositories.py Normal file
View File

@@ -0,0 +1,105 @@
from datetime import date
from decimal import Decimal
from grid_trading.domain.models import Account, Instrument, StrategyTemplate, Trade, TradeGroup, TradeSide
from grid_trading.storage.repositories import Repository
def test_database_round_trips_account_instrument_template_and_trade(tmp_path):
db_path = tmp_path / "grid.db"
repo = Repository(db_path)
repo.initialize()
account = repo.save_account(Account(id=None, name="主账户", initial_cash=Decimal("100000"), notes="first"))
template = repo.save_strategy_template(
StrategyTemplate(
id=None,
name="默认模板",
grid_spacing_pct=Decimal("0.03"),
amount_per_grid=Decimal("5000"),
base_target_amount=Decimal("20000"),
max_position_amount=Decimal("80000"),
min_lot=100,
is_default=True,
)
)
instrument = repo.save_instrument(
Instrument(
id=None,
code="510300",
name="沪深300ETF",
market="ETF",
lot_size=100,
manual_price=Decimal("3.95"),
)
)
trade = repo.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 8),
side=TradeSide.BUY,
price=Decimal("3.90"),
quantity=1000,
commission=Decimal("5"),
transfer_fee=Decimal("0.04"),
trade_group=TradeGroup.GRID,
notes="first buy",
)
)
repo.close()
reopened = Repository(db_path)
reopened.initialize()
assert reopened.list_accounts() == [account]
assert reopened.get_default_strategy_template() == template
assert reopened.list_instruments() == [instrument]
assert reopened.list_trades(account_id=account.id) == [trade]
def test_trade_update_and_delete_are_persistent(tmp_path):
repo = Repository(tmp_path / "grid.db")
repo.initialize()
account = repo.save_account(Account(id=None, name="主账户", initial_cash=Decimal("50000")))
instrument = repo.save_instrument(Instrument(id=None, code="600000", name="浦发银行"))
trade = repo.save_trade(
Trade(
id=None,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("10"),
quantity=100,
commission=Decimal("5"),
trade_group=TradeGroup.BASE,
)
)
updated = repo.save_trade(
Trade(
id=trade.id,
account_id=account.id,
instrument_id=instrument.id,
trade_date=date(2026, 7, 7),
side=TradeSide.BUY,
price=Decimal("9.8"),
quantity=200,
commission=Decimal("5"),
trade_group=TradeGroup.BASE,
notes="corrected",
)
)
assert repo.get_trade(trade.id) == updated
repo.close()
reopened = Repository(tmp_path / "grid.db")
reopened.initialize()
assert reopened.get_trade(trade.id) == updated
reopened.delete_trade(trade.id)
assert reopened.get_trade(trade.id) is None
assert reopened.list_trades(account_id=account.id) == []