fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: 串行化持仓账本写入以消除并发超售窗口 (#742) (#743)

Browse Source 
* fix: serialize portfolio ledger writes to close oversell race

* test: cover busy responses for portfolio event APIs

---------

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>
This commit is contained in:
Alfred
2026-03-18 19:10:15 +08:00
committed by GitHub
parent 216a77f6a2
commit f57f21fdaa
11 changed files with 780 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -53,7 +53,7 @@
> Web 管理认证支持运行时开关；如果系统中已保留管理员密码，重新开启认证时必须提供当前密码，避免在认证关闭窗口内直接获取新的管理员会话。
> 多进程/多 worker 部署时，认证开关仅在当前进程即时生效；需重启或滚动重启全部 worker 以统一状态。
> 持仓管理补充说明：卖出录入现在会在写入前校验可用持仓，超售会直接拒绝；如果历史里误录了交易 / 资金流水 / 公司行为，可在 Web `/portfolio` 页的事件列表中直接删除后恢复快照。
> 持仓管理补充说明：卖出录入现在会在写入前校验可用持仓，超售会直接拒绝；如果历史里误录了交易 / 资金流水 / 公司行为，可在 Web `/portfolio` 页的事件列表中直接删除后恢复快照。高并发写入场景下，直接持仓写接口可能返回 `409 portfolio_busy`，提示账本正在处理另一笔变更；CSV 导入仍保持逐条提交与部分成功语义。
### 技术栈与数据来源

23
api/v1/endpoints/portfolio.py
View File

@@ -34,6 +34,7 @@ from api.v1.schemas.portfolio import (
from src.services.portfolio_import_service import PortfolioImportService
from src.services.portfolio_risk_service import PortfolioRiskService
from src.services.portfolio_service import (
PortfolioBusyError,
PortfolioConflictError,
PortfolioOversellError,
PortfolioService,
@@ -193,6 +194,8 @@ def create_trade(request: PortfolioTradeCreateRequest) -> PortfolioEventCreatedR
note=request.note,
)
return PortfolioEventCreatedResponse(**data)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except PortfolioOversellError as exc:
raise _conflict_error(error="portfolio_oversell", message=str(exc))
except PortfolioConflictError as exc:
@@ -239,7 +242,7 @@ def list_trades(
@router.delete(
"/trades/{trade_id}",
response_model=PortfolioDeleteResponse,
responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
responses={404: {"model": ErrorResponse}, 409: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
summary="Delete trade event",
)
def delete_trade(trade_id: int) -> PortfolioDeleteResponse:
@@ -252,6 +255,8 @@ def delete_trade(trade_id: int) -> PortfolioDeleteResponse:
detail={"error": "not_found", "message": f"Trade not found: {trade_id}"},
)
return PortfolioDeleteResponse(deleted=1)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except HTTPException:
raise
except Exception as exc:
@@ -261,7 +266,7 @@ def delete_trade(trade_id: int) -> PortfolioDeleteResponse:
@router.post(
"/cash-ledger",
response_model=PortfolioEventCreatedResponse,
responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
responses={400: {"model": ErrorResponse}, 409: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
summary="Record cash event",
)
def create_cash_ledger(request: PortfolioCashLedgerCreateRequest) -> PortfolioEventCreatedResponse:
@@ -276,6 +281,8 @@ def create_cash_ledger(request: PortfolioCashLedgerCreateRequest) -> PortfolioEv
note=request.note,
)
return PortfolioEventCreatedResponse(**data)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except ValueError as exc:
raise _bad_request(exc)
except Exception as exc:
@@ -316,7 +323,7 @@ def list_cash_ledger(
@router.delete(
"/cash-ledger/{entry_id}",
response_model=PortfolioDeleteResponse,
responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
responses={404: {"model": ErrorResponse}, 409: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
summary="Delete cash ledger event",
)
def delete_cash_ledger(entry_id: int) -> PortfolioDeleteResponse:
@@ -329,6 +336,8 @@ def delete_cash_ledger(entry_id: int) -> PortfolioDeleteResponse:
detail={"error": "not_found", "message": f"Cash ledger entry not found: {entry_id}"},
)
return PortfolioDeleteResponse(deleted=1)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except HTTPException:
raise
except Exception as exc:
@@ -338,7 +347,7 @@ def delete_cash_ledger(entry_id: int) -> PortfolioDeleteResponse:
@router.post(
"/corporate-actions",
response_model=PortfolioEventCreatedResponse,
responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
responses={400: {"model": ErrorResponse}, 409: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
summary="Record corporate action event",
)
def create_corporate_action(request: PortfolioCorporateActionCreateRequest) -> PortfolioEventCreatedResponse:
@@ -356,6 +365,8 @@ def create_corporate_action(request: PortfolioCorporateActionCreateRequest) -> P
note=request.note,
)
return PortfolioEventCreatedResponse(**data)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except ValueError as exc:
raise _bad_request(exc)
except Exception as exc:
@@ -398,7 +409,7 @@ def list_corporate_actions(
@router.delete(
"/corporate-actions/{action_id}",
response_model=PortfolioDeleteResponse,
responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
responses={404: {"model": ErrorResponse}, 409: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
summary="Delete corporate action event",
)
def delete_corporate_action(action_id: int) -> PortfolioDeleteResponse:
@@ -411,6 +422,8 @@ def delete_corporate_action(action_id: int) -> PortfolioDeleteResponse:
detail={"error": "not_found", "message": f"Corporate action not found: {action_id}"},
)
return PortfolioDeleteResponse(deleted=1)
except PortfolioBusyError as exc:
raise _conflict_error(error="portfolio_busy", message=str(exc))
except HTTPException:
raise
except Exception as exc:

11
apps/dsa-web/src/api/error.ts
View File

@@ -7,6 +7,7 @@ export type ApiErrorCategory =
| 'model_tool_incompatible'
| 'invalid_tool_call'
| 'portfolio_oversell'
| 'portfolio_busy'
| 'upstream_llm_400'
| 'upstream_timeout'
| 'upstream_network'
@@ -332,6 +333,16 @@ export function parseApiError(error: unknown): ParsedApiError {
});
}
if (errorCode === 'portfolio_busy' || includesAny(matchText, ['portfolio ledger is busy'])) {
return createParsedApiError({
title: '持仓账本正忙',
message: '持仓账本正在处理另一笔变更，请稍后重试。',
rawMessage,
status,
category: 'portfolio_busy',
});
}
const noConfiguredLlm = (
includesAny(matchText, ['all llm models failed']) && includesAny(matchText, ['last error: none'])
) || includesAny(matchText, [

1
docs/CHANGELOG.md
View File

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
### 说明
- 💼 **持仓账本并发写入串行化**（#742）— 持仓源事件写入/删除现在会在 SQLite 下先获取串行化写锁，减少并发卖出把超售流水写入账本的窗口；直接持仓写接口在锁竞争时返回 `409 portfolio_busy`，CSV 导入保持逐条提交并把 busy 计入 `failed_count`。
- 🚀 **Agent 与普通分析模型解耦（Issue #692）** — 新增 `AGENT_LITELLM_MODEL`（留空继承 `LITELLM_MODEL`，无前缀按 `openai/<model>` 归一）；Agent 执行链路与 `/api/v1/agent/models` 的 `is_primary/is_fallback` 标记改为基于 Agent 实际模型链路；系统配置与启动期校验补齐 `AGENT_LITELLM_MODEL` 的 `unknown_model/missing_runtime_source` 检查；Web 设置页新增 Agent 主模型选择并与渠道模式运行时配置同步。
## [3.8.0] - 2026-03-17

2
docs/full-guide.md
View File

@@ -940,6 +940,7 @@ A: 检查是否启用了 Actions，以及 cron 表达式是否正确（注意是
### Error and stability semantics
- `trade_uid` unique conflict returns `409` (API conflict semantics).
- sell writes now validate available quantity before insert; oversell is rejected with `409 portfolio_oversell`.
- portfolio source-event writes now serialize through a SQLite write lock; direct write/delete endpoints may return `409 portfolio_busy` when another ledger mutation is in progress.
- Snapshot write path is atomic for positions/lots/daily snapshot.
- FX conversion keeps fail-open behavior (fallback 1:1 with stale marker) to avoid pipeline interruption.
@@ -955,6 +956,7 @@ A: 检查是否启用了 Actions，以及 cron 表达式是否正确（注意是
### CSV import
- Supported broker ids: `huatai`, `citic`, `cmb`.
- Unified workflow: parse CSV into normalized records, then commit into portfolio trades.
- Commit remains row-by-row instead of one long transaction; busy rows count into `failed_count` rather than converting the whole request to `409`.
- Dedup policy:
- First key: `trade_uid` (account-scoped)
- Fallback key: deterministic hash of date/symbol/side/qty/price/fee/tax/currency

492
src/repositories/portfolio_repo.py
View File

@@ -7,11 +7,12 @@ Provides DB access helpers for portfolio account/events/snapshot tables.
from __future__ import annotations
import logging
from contextlib import contextmanager
from datetime import date, datetime
from typing import Any, Dict, Iterable, List, Optional, Tuple
from sqlalchemy import and_, delete, desc, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import IntegrityError, OperationalError
from src.storage import (
DatabaseManager,
@@ -37,6 +38,10 @@ class DuplicateTradeDedupHashError(Exception):
"""Raised when dedup hash conflicts with existing record in one account."""
class PortfolioBusyError(Exception):
"""Raised when SQLite write serialization cannot acquire the ledger lock."""
class PortfolioRepository:
"""DB access layer for portfolio P0 domain."""
@@ -71,12 +76,11 @@ class PortfolioRepository:
def get_account(self, account_id: int, include_inactive: bool = False) -> Optional[PortfolioAccount]:
with self.db.get_session() as session:
conditions = [PortfolioAccount.id == account_id]
if not include_inactive:
conditions.append(PortfolioAccount.is_active.is_(True))
return session.execute(
select(PortfolioAccount).where(and_(*conditions)).limit(1)
).scalar_one_or_none()
return self.get_account_in_session(
session=session,
account_id=account_id,
include_inactive=include_inactive,
)
def list_accounts(self, include_inactive: bool = False) -> List[PortfolioAccount]:
with self.db.get_session() as session:
@@ -86,6 +90,20 @@ class PortfolioRepository:
rows = session.execute(query.order_by(PortfolioAccount.id.asc())).scalars().all()
return list(rows)
def get_account_in_session(
self,
*,
session: Any,
account_id: int,
include_inactive: bool = False,
) -> Optional[PortfolioAccount]:
conditions = [PortfolioAccount.id == account_id]
if not include_inactive:
conditions.append(PortfolioAccount.is_active.is_(True))
return session.execute(
select(PortfolioAccount).where(and_(*conditions)).limit(1)
).scalar_one_or_none()
def update_account(self, account_id: int, fields: Dict[str, Any]) -> Optional[PortfolioAccount]:
with self.db.get_session() as session:
row = session.execute(
@@ -115,6 +133,31 @@ class PortfolioRepository:
# ------------------------------------------------------------------
# Event writes
# ------------------------------------------------------------------
@contextmanager
def portfolio_write_session(self):
session = self.db.get_session()
try:
session.connection().exec_driver_sql("BEGIN IMMEDIATE")
except OperationalError as exc:
session.close()
if self._is_sqlite_locked_error(exc):
raise PortfolioBusyError("Portfolio ledger is busy; please retry shortly.") from exc
raise
try:
yield session
session.commit()
except OperationalError as exc:
session.rollback()
if self._is_sqlite_locked_error(exc):
raise PortfolioBusyError("Portfolio ledger is busy; please retry shortly.") from exc
raise
except Exception:
session.rollback()
raise
finally:
session.close()
def add_trade(
self,
*,
@@ -132,8 +175,9 @@ class PortfolioRepository:
note: Optional[str] = None,
dedup_hash: Optional[str] = None,
) -> PortfolioTrade:
with self.db.get_session() as session:
row = PortfolioTrade(
with self.portfolio_write_session() as session:
row = self.add_trade_in_session(
session=session,
account_id=account_id,
trade_uid=trade_uid,
symbol=symbol,
@@ -148,31 +192,7 @@ class PortfolioRepository:
note=note,
dedup_hash=dedup_hash,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=trade_date,
)
try:
session.commit()
except IntegrityError as exc:
session.rollback()
err_text = str(getattr(exc, "orig", exc)).lower()
if trade_uid and ("uix_portfolio_trade_uid" in err_text or "unique" in err_text):
raise DuplicateTradeUidError(
f"Duplicate trade_uid for account_id={account_id}: {trade_uid}"
) from exc
if dedup_hash and (
"uix_portfolio_trade_dedup_hash" in err_text
or "portfolio_trades.account_id, portfolio_trades.dedup_hash" in err_text
or ("unique" in err_text and "dedup_hash" in err_text)
):
raise DuplicateTradeDedupHashError(
f"Duplicate dedup_hash for account_id={account_id}: {dedup_hash}"
) from exc
raise
session.refresh(row)
session.expunge(row)
return row
def add_cash_ledger(
@@ -185,8 +205,9 @@ class PortfolioRepository:
currency: str,
note: Optional[str] = None,
) -> PortfolioCashLedger:
with self.db.get_session() as session:
row = PortfolioCashLedger(
with self.portfolio_write_session() as session:
row = self.add_cash_ledger_in_session(
session=session,
account_id=account_id,
event_date=event_date,
direction=direction,
@@ -194,14 +215,7 @@ class PortfolioRepository:
currency=currency,
note=note,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=event_date,
)
session.commit()
session.refresh(row)
session.expunge(row)
return row
def add_corporate_action(
@@ -217,8 +231,9 @@ class PortfolioRepository:
split_ratio: Optional[float] = None,
note: Optional[str] = None,
) -> PortfolioCorporateAction:
with self.db.get_session() as session:
row = PortfolioCorporateAction(
with self.portfolio_write_session() as session:
row = self.add_corporate_action_in_session(
session=session,
account_id=account_id,
symbol=symbol,
market=market,
@@ -229,63 +244,20 @@ class PortfolioRepository:
split_ratio=split_ratio,
note=note,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=effective_date,
)
session.commit()
session.refresh(row)
session.expunge(row)
return row
def delete_trade(self, trade_id: int) -> bool:
with self.db.get_session() as session:
row = session.execute(
select(PortfolioTrade).where(PortfolioTrade.id == trade_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.trade_date,
)
session.delete(row)
session.commit()
return True
with self.portfolio_write_session() as session:
return self.delete_trade_in_session(session=session, trade_id=trade_id)
def delete_cash_ledger(self, entry_id: int) -> bool:
with self.db.get_session() as session:
row = session.execute(
select(PortfolioCashLedger).where(PortfolioCashLedger.id == entry_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.event_date,
)
session.delete(row)
session.commit()
return True
with self.portfolio_write_session() as session:
return self.delete_cash_ledger_in_session(session=session, entry_id=entry_id)
def delete_corporate_action(self, action_id: int) -> bool:
with self.db.get_session() as session:
row = session.execute(
select(PortfolioCorporateAction).where(PortfolioCorporateAction.id == action_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.effective_date,
)
session.delete(row)
session.commit()
return True
with self.portfolio_write_session() as session:
return self.delete_corporate_action_in_session(session=session, action_id=action_id)
def has_trade_uid(self, account_id: int, trade_uid: Optional[str]) -> bool:
"""Return True when trade_uid already exists in the account."""
@@ -293,15 +265,7 @@ class PortfolioRepository:
if not uid:
return False
with self.db.get_session() as session:
row = session.execute(
select(PortfolioTrade.id).where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.trade_uid == uid,
)
).limit(1)
).scalar_one_or_none()
return row is not None
return self.has_trade_uid_in_session(session=session, account_id=account_id, trade_uid=uid)
def has_trade_dedup_hash(self, account_id: int, dedup_hash: Optional[str]) -> bool:
"""Return True when dedup hash already exists in the account."""
@@ -309,60 +273,265 @@ class PortfolioRepository:
if not hash_value:
return False
with self.db.get_session() as session:
row = session.execute(
select(PortfolioTrade.id).where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.dedup_hash == hash_value,
)
).limit(1)
).scalar_one_or_none()
return row is not None
return self.has_trade_dedup_hash_in_session(
session=session,
account_id=account_id,
dedup_hash=hash_value,
)
def has_trade_uid_in_session(self, *, session: Any, account_id: int, trade_uid: str) -> bool:
row = session.execute(
select(PortfolioTrade.id).where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.trade_uid == trade_uid,
)
).limit(1)
).scalar_one_or_none()
return row is not None
def has_trade_dedup_hash_in_session(self, *, session: Any, account_id: int, dedup_hash: str) -> bool:
row = session.execute(
select(PortfolioTrade.id).where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.dedup_hash == dedup_hash,
)
).limit(1)
).scalar_one_or_none()
return row is not None
def add_trade_in_session(
self,
*,
session: Any,
account_id: int,
trade_uid: Optional[str],
symbol: str,
market: str,
currency: str,
trade_date: date,
side: str,
quantity: float,
price: float,
fee: float,
tax: float,
note: Optional[str] = None,
dedup_hash: Optional[str] = None,
) -> PortfolioTrade:
row = PortfolioTrade(
account_id=account_id,
trade_uid=trade_uid,
symbol=symbol,
market=market,
currency=currency,
trade_date=trade_date,
side=side,
quantity=quantity,
price=price,
fee=fee,
tax=tax,
note=note,
dedup_hash=dedup_hash,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=trade_date,
)
try:
session.flush()
except IntegrityError as exc:
raise self._translate_trade_integrity_error(
exc=exc,
account_id=account_id,
trade_uid=trade_uid,
dedup_hash=dedup_hash,
) from exc
session.refresh(row)
return row
def add_cash_ledger_in_session(
self,
*,
session: Any,
account_id: int,
event_date: date,
direction: str,
amount: float,
currency: str,
note: Optional[str] = None,
) -> PortfolioCashLedger:
row = PortfolioCashLedger(
account_id=account_id,
event_date=event_date,
direction=direction,
amount=amount,
currency=currency,
note=note,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=event_date,
)
session.flush()
session.refresh(row)
return row
def add_corporate_action_in_session(
self,
*,
session: Any,
account_id: int,
symbol: str,
market: str,
currency: str,
effective_date: date,
action_type: str,
cash_dividend_per_share: Optional[float] = None,
split_ratio: Optional[float] = None,
note: Optional[str] = None,
) -> PortfolioCorporateAction:
row = PortfolioCorporateAction(
account_id=account_id,
symbol=symbol,
market=market,
currency=currency,
effective_date=effective_date,
action_type=action_type,
cash_dividend_per_share=cash_dividend_per_share,
split_ratio=split_ratio,
note=note,
)
session.add(row)
self._invalidate_account_cache_in_session(
session=session,
account_id=account_id,
from_date=effective_date,
)
session.flush()
session.refresh(row)
return row
def delete_trade_in_session(self, *, session: Any, trade_id: int) -> bool:
row = session.execute(
select(PortfolioTrade).where(PortfolioTrade.id == trade_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.trade_date,
)
session.delete(row)
session.flush()
return True
def delete_cash_ledger_in_session(self, *, session: Any, entry_id: int) -> bool:
row = session.execute(
select(PortfolioCashLedger).where(PortfolioCashLedger.id == entry_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.event_date,
)
session.delete(row)
session.flush()
return True
def delete_corporate_action_in_session(self, *, session: Any, action_id: int) -> bool:
row = session.execute(
select(PortfolioCorporateAction).where(PortfolioCorporateAction.id == action_id).limit(1)
).scalar_one_or_none()
if row is None:
return False
self._invalidate_account_cache_in_session(
session=session,
account_id=int(row.account_id),
from_date=row.effective_date,
)
session.delete(row)
session.flush()
return True
# ------------------------------------------------------------------
# Event reads
# ------------------------------------------------------------------
def list_trades(self, account_id: int, as_of: date) -> List[PortfolioTrade]:
with self.db.get_session() as session:
rows = session.execute(
select(PortfolioTrade)
.where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.trade_date <= as_of,
)
return self.list_trades_in_session(session=session, account_id=account_id, as_of=as_of)
def list_trades_in_session(
self,
*,
session: Any,
account_id: int,
as_of: date,
) -> List[PortfolioTrade]:
rows = session.execute(
select(PortfolioTrade)
.where(
and_(
PortfolioTrade.account_id == account_id,
PortfolioTrade.trade_date <= as_of,
)
.order_by(PortfolioTrade.trade_date.asc(), PortfolioTrade.id.asc())
).scalars().all()
return list(rows)
)
.order_by(PortfolioTrade.trade_date.asc(), PortfolioTrade.id.asc())
).scalars().all()
return list(rows)
def list_cash_ledger(self, account_id: int, as_of: date) -> List[PortfolioCashLedger]:
with self.db.get_session() as session:
rows = session.execute(
select(PortfolioCashLedger)
.where(
and_(
PortfolioCashLedger.account_id == account_id,
PortfolioCashLedger.event_date <= as_of,
)
return self.list_cash_ledger_in_session(session=session, account_id=account_id, as_of=as_of)
def list_cash_ledger_in_session(
self,
*,
session: Any,
account_id: int,
as_of: date,
) -> List[PortfolioCashLedger]:
rows = session.execute(
select(PortfolioCashLedger)
.where(
and_(
PortfolioCashLedger.account_id == account_id,
PortfolioCashLedger.event_date <= as_of,
)
.order_by(PortfolioCashLedger.event_date.asc(), PortfolioCashLedger.id.asc())
).scalars().all()
return list(rows)
)
.order_by(PortfolioCashLedger.event_date.asc(), PortfolioCashLedger.id.asc())
).scalars().all()
return list(rows)
def list_corporate_actions(self, account_id: int, as_of: date) -> List[PortfolioCorporateAction]:
with self.db.get_session() as session:
rows = session.execute(
select(PortfolioCorporateAction)
.where(
and_(
PortfolioCorporateAction.account_id == account_id,
PortfolioCorporateAction.effective_date <= as_of,
)
return self.list_corporate_actions_in_session(session=session, account_id=account_id, as_of=as_of)
def list_corporate_actions_in_session(
self,
*,
session: Any,
account_id: int,
as_of: date,
) -> List[PortfolioCorporateAction]:
rows = session.execute(
select(PortfolioCorporateAction)
.where(
and_(
PortfolioCorporateAction.account_id == account_id,
PortfolioCorporateAction.effective_date <= as_of,
)
.order_by(PortfolioCorporateAction.effective_date.asc(), PortfolioCorporateAction.id.asc())
).scalars().all()
return list(rows)
)
.order_by(PortfolioCorporateAction.effective_date.asc(), PortfolioCorporateAction.id.asc())
).scalars().all()
return list(rows)
def get_first_activity_date(self, *, account_id: int, as_of: date) -> Optional[date]:
"""Return earliest event date (trade/cash/corporate action) for one account."""
@@ -705,6 +874,41 @@ class PortfolioRepository:
)
)
@staticmethod
def _is_sqlite_locked_error(exc: OperationalError) -> bool:
err_text = str(getattr(exc, "orig", exc)).lower()
return any(
token in err_text
for token in (
"database is locked",
"database schema is locked",
"database table is locked",
)
)
@staticmethod
def _translate_trade_integrity_error(
*,
exc: IntegrityError,
account_id: int,
trade_uid: Optional[str],
dedup_hash: Optional[str],
) -> Exception:
err_text = str(getattr(exc, "orig", exc)).lower()
if trade_uid and ("uix_portfolio_trade_uid" in err_text or "unique" in err_text):
return DuplicateTradeUidError(
f"Duplicate trade_uid for account_id={account_id}: {trade_uid}"
)
if dedup_hash and (
"uix_portfolio_trade_dedup_hash" in err_text
or "portfolio_trades.account_id, portfolio_trades.dedup_hash" in err_text
or ("unique" in err_text and "dedup_hash" in err_text)
):
return DuplicateTradeDedupHashError(
f"Duplicate dedup_hash for account_id={account_id}: {dedup_hash}"
)
return exc
def upsert_daily_snapshot(
self,
*,

10
src/services/portfolio_import_service.py
View File

@@ -14,7 +14,12 @@ import pandas as pd
from data_provider.base import canonical_stock_code
from src.repositories.portfolio_repo import PortfolioRepository
from src.services.portfolio_service import PortfolioConflictError, PortfolioOversellError, PortfolioService
from src.services.portfolio_service import (
PortfolioBusyError,
PortfolioConflictError,
PortfolioOversellError,
PortfolioService,
)
logger = logging.getLogger(__name__)
@@ -247,6 +252,9 @@ class PortfolioImportService:
except PortfolioOversellError as exc:
failed_count += 1
errors.append(f"idx={i}: {exc}")
except PortfolioBusyError as exc:
failed_count += 1
errors.append(f"idx={i}: portfolio_busy: {exc}")
except Exception as exc:
failed_count += 1
errors.append(f"idx={i}: {exc}")

198
src/services/portfolio_service.py
View File

@@ -15,11 +15,14 @@ from src.config import get_config
from src.repositories.portfolio_repo import (
DuplicateTradeDedupHashError,
DuplicateTradeUidError,
PortfolioBusyError as RepoPortfolioBusyError,
PortfolioRepository,
)
logger = logging.getLogger(__name__)
PortfolioBusyError = RepoPortfolioBusyError
try:
import yfinance as yf
except Exception: # pragma: no cover - optional dependency path
@@ -160,7 +163,6 @@ class PortfolioService:
dedup_hash: Optional[str] = None,
note: Optional[str] = None,
) -> Dict[str, Any]:
account = self._require_active_account(account_id)
side_norm = (side or "").strip().lower()
if side_norm not in VALID_SIDES:
raise ValueError("side must be buy or sell")
@@ -168,48 +170,51 @@ class PortfolioService:
raise ValueError("quantity and price must be > 0")
if fee < 0 or tax < 0:
raise ValueError("fee and tax must be >= 0")
market_norm = self._normalize_market(market or account.market)
currency_norm = self._normalize_currency(currency or self._default_currency_for_market(market_norm))
symbol_norm = canonical_stock_code(symbol)
if not symbol_norm:
raise ValueError("symbol is required")
trade_uid_norm = (trade_uid or "").strip() or None
dedup_hash_norm = (dedup_hash or "").strip() or None
self._validate_trade_identity(
account_id=account_id,
trade_uid=trade_uid_norm,
dedup_hash=dedup_hash_norm,
)
if side_norm == "sell":
self._validate_sell_quantity(
account_id=account_id,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
trade_date=trade_date,
quantity=float(quantity),
)
try:
row = self.repo.add_trade(
account_id=account_id,
trade_uid=trade_uid_norm,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
trade_date=trade_date,
side=side_norm,
quantity=float(quantity),
price=float(price),
fee=float(fee),
tax=float(tax),
note=(note or "").strip() or None,
dedup_hash=dedup_hash_norm,
)
with self.repo.portfolio_write_session() as session:
account = self._require_active_account_in_session(session=session, account_id=account_id)
market_norm = self._normalize_market(market or account.market)
currency_norm = self._normalize_currency(currency or self._default_currency_for_market(market_norm))
self._validate_trade_identity(
account_id=account_id,
trade_uid=trade_uid_norm,
dedup_hash=dedup_hash_norm,
session=session,
)
if side_norm == "sell":
self._validate_sell_quantity(
account_id=account_id,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
trade_date=trade_date,
quantity=float(quantity),
session=session,
)
row = self.repo.add_trade_in_session(
session=session,
account_id=account_id,
trade_uid=trade_uid_norm,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
trade_date=trade_date,
side=side_norm,
quantity=float(quantity),
price=float(price),
fee=float(fee),
tax=float(tax),
note=(note or "").strip() or None,
dedup_hash=dedup_hash_norm,
)
return {"id": int(row.id)}
except (DuplicateTradeUidError, DuplicateTradeDedupHashError) as exc:
raise PortfolioConflictError(str(exc)) from exc
return {"id": row.id}
def record_cash_ledger(
self,
@@ -221,22 +226,24 @@ class PortfolioService:
currency: Optional[str] = None,
note: Optional[str] = None,
) -> Dict[str, Any]:
account = self._require_active_account(account_id)
direction_norm = (direction or "").strip().lower()
if direction_norm not in VALID_CASH_DIRECTIONS:
raise ValueError("direction must be in or out")
if amount <= 0:
raise ValueError("amount must be > 0")
currency_norm = self._normalize_currency(currency or account.base_currency)
row = self.repo.add_cash_ledger(
account_id=account_id,
event_date=event_date,
direction=direction_norm,
amount=float(amount),
currency=currency_norm,
note=(note or "").strip() or None,
)
return {"id": row.id}
with self.repo.portfolio_write_session() as session:
account = self._require_active_account_in_session(session=session, account_id=account_id)
currency_norm = self._normalize_currency(currency or account.base_currency)
row = self.repo.add_cash_ledger_in_session(
session=session,
account_id=account_id,
event_date=event_date,
direction=direction_norm,
amount=float(amount),
currency=currency_norm,
note=(note or "").strip() or None,
)
return {"id": int(row.id)}
def record_corporate_action(
self,
@@ -251,45 +258,48 @@ class PortfolioService:
split_ratio: Optional[float] = None,
note: Optional[str] = None,
) -> Dict[str, Any]:
account = self._require_active_account(account_id)
action_type_norm = (action_type or "").strip().lower()
if action_type_norm not in VALID_CORPORATE_ACTIONS:
raise ValueError("action_type must be cash_dividend or split_adjustment")
market_norm = self._normalize_market(market or account.market)
currency_norm = self._normalize_currency(currency or self._default_currency_for_market(market_norm))
symbol_norm = canonical_stock_code(symbol)
if not symbol_norm:
raise ValueError("symbol is required")
if action_type_norm == "cash_dividend":
if cash_dividend_per_share is None or cash_dividend_per_share < 0:
raise ValueError("cash_dividend_per_share must be >= 0 for cash_dividend")
if action_type_norm == "split_adjustment":
if split_ratio is None or split_ratio <= 0:
raise ValueError("split_ratio must be > 0 for split_adjustment")
row = self.repo.add_corporate_action(
account_id=account_id,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
effective_date=effective_date,
action_type=action_type_norm,
cash_dividend_per_share=cash_dividend_per_share,
split_ratio=split_ratio,
note=(note or "").strip() or None,
)
return {"id": row.id}
with self.repo.portfolio_write_session() as session:
account = self._require_active_account_in_session(session=session, account_id=account_id)
market_norm = self._normalize_market(market or account.market)
currency_norm = self._normalize_currency(currency or self._default_currency_for_market(market_norm))
symbol_norm = canonical_stock_code(symbol)
if not symbol_norm:
raise ValueError("symbol is required")
row = self.repo.add_corporate_action_in_session(
session=session,
account_id=account_id,
symbol=symbol_norm,
market=market_norm,
currency=currency_norm,
effective_date=effective_date,
action_type=action_type_norm,
cash_dividend_per_share=cash_dividend_per_share,
split_ratio=split_ratio,
note=(note or "").strip() or None,
)
return {"id": int(row.id)}
def delete_trade_event(self, trade_id: int) -> bool:
return self.repo.delete_trade(trade_id)
with self.repo.portfolio_write_session() as session:
return self.repo.delete_trade_in_session(session=session, trade_id=trade_id)
def delete_cash_ledger_event(self, entry_id: int) -> bool:
return self.repo.delete_cash_ledger(entry_id)
with self.repo.portfolio_write_session() as session:
return self.repo.delete_cash_ledger_in_session(session=session, entry_id=entry_id)
def delete_corporate_action_event(self, action_id: int) -> bool:
return self.repo.delete_corporate_action(action_id)
with self.repo.portfolio_write_session() as session:
return self.repo.delete_corporate_action_in_session(session=session, action_id=action_id)
def list_trade_events(
self,
@@ -590,10 +600,11 @@ class PortfolioService:
account_id: int,
trade_uid: Optional[str],
dedup_hash: Optional[str],
session: Optional[Any] = None,
) -> None:
if trade_uid and self.repo.has_trade_uid(account_id, trade_uid):
if trade_uid and self._has_trade_uid(account_id=account_id, trade_uid=trade_uid, session=session):
raise PortfolioConflictError(f"Duplicate trade_uid for account_id={account_id}: {trade_uid}")
if dedup_hash and self.repo.has_trade_dedup_hash(account_id, dedup_hash):
if dedup_hash and self._has_trade_dedup_hash(account_id=account_id, dedup_hash=dedup_hash, session=session):
raise PortfolioConflictError(f"Duplicate dedup_hash for account_id={account_id}: {dedup_hash}")
def _validate_sell_quantity(
@@ -605,6 +616,7 @@ class PortfolioService:
currency: str,
trade_date: date,
quantity: float,
session: Optional[Any] = None,
) -> None:
key = (
canonical_stock_code(symbol),
@@ -615,6 +627,7 @@ class PortfolioService:
account_id=account_id,
key=key,
as_of_date=trade_date,
session=session,
)
if available_quantity + EPS < quantity:
raise PortfolioOversellError(
@@ -630,9 +643,18 @@ class PortfolioService:
account_id: int,
key: Tuple[str, str, str],
as_of_date: date,
session: Optional[Any] = None,
) -> float:
trades = self.repo.list_trades(account_id, as_of=as_of_date)
corporate_actions = self.repo.list_corporate_actions(account_id, as_of=as_of_date)
if session is None:
trades = self.repo.list_trades(account_id, as_of=as_of_date)
corporate_actions = self.repo.list_corporate_actions(account_id, as_of=as_of_date)
else:
trades = self.repo.list_trades_in_session(session=session, account_id=account_id, as_of=as_of_date)
corporate_actions = self.repo.list_corporate_actions_in_session(
session=session,
account_id=account_id,
as_of=as_of_date,
)
events = []
for row in corporate_actions:
@@ -1211,6 +1233,36 @@ class PortfolioService:
raise ValueError(f"Active account not found: {account_id}")
return account
def _require_active_account_in_session(self, *, session: Any, account_id: int) -> Any:
account = self.repo.get_account_in_session(
session=session,
account_id=account_id,
include_inactive=False,
)
if account is None:
raise ValueError(f"Active account not found: {account_id}")
return account
def _has_trade_uid(self, *, account_id: int, trade_uid: str, session: Optional[Any] = None) -> bool:
if session is None:
return self.repo.has_trade_uid(account_id, trade_uid)
return self.repo.has_trade_uid_in_session(session=session, account_id=account_id, trade_uid=trade_uid)
def _has_trade_dedup_hash(
self,
*,
account_id: int,
dedup_hash: str,
session: Optional[Any] = None,
) -> bool:
if session is None:
return self.repo.has_trade_dedup_hash(account_id, dedup_hash)
return self.repo.has_trade_dedup_hash_in_session(
session=session,
account_id=account_id,
dedup_hash=dedup_hash,
)
@staticmethod
def _account_to_dict(row: Any) -> Dict[str, Any]:
return {

103
tests/test_portfolio_api.py
View File

@@ -9,7 +9,7 @@ import tempfile
import unittest
from datetime import date
from pathlib import Path
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
import pandas as pd
from fastapi.testclient import TestClient
@@ -23,6 +23,7 @@ except ModuleNotFoundError:
import src.auth as auth
from api.app import create_app
from src.config import Config
from src.services.portfolio_service import PortfolioBusyError
from src.storage import DatabaseManager
@@ -432,6 +433,106 @@ class PortfolioApiTestCase(unittest.TestCase):
missing_trade = self.client.delete("/api/v1/portfolio/trades/999999")
self.assertEqual(missing_trade.status_code, 404)
def test_create_trade_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.record_trade",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.post(
"/api/v1/portfolio/trades",
json={
"account_id": 1,
"symbol": "600519",
"trade_date": "2026-01-02",
"side": "buy",
"quantity": 10,
"price": 100,
"fee": 0,
"tax": 0,
"market": "cn",
"currency": "CNY",
},
)
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_delete_trade_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.delete_trade_event",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.delete("/api/v1/portfolio/trades/1")
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_create_cash_ledger_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.record_cash_ledger",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.post(
"/api/v1/portfolio/cash-ledger",
json={
"account_id": 1,
"event_date": "2026-01-02",
"direction": "in",
"amount": 1000,
"currency": "CNY",
},
)
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_delete_cash_ledger_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.delete_cash_ledger_event",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.delete("/api/v1/portfolio/cash-ledger/1")
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_create_corporate_action_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.record_corporate_action",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.post(
"/api/v1/portfolio/corporate-actions",
json={
"account_id": 1,
"symbol": "600519",
"effective_date": "2026-01-02",
"action_type": "split_adjustment",
"market": "cn",
"currency": "CNY",
"split_ratio": 2.0,
},
)
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_delete_corporate_action_busy_returns_409(self) -> None:
with patch(
"api.v1.endpoints.portfolio.PortfolioService.delete_corporate_action_event",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
resp = self.client.delete("/api/v1/portfolio/corporate-actions/1")
self.assertEqual(resp.status_code, 409)
detail = resp.json()
self.assertEqual(detail.get("error"), "portfolio_busy")
def test_csv_broker_list_endpoint(self) -> None:
resp = self.client.get("/api/v1/portfolio/imports/csv/brokers")
self.assertEqual(resp.status_code, 200)

36
tests/test_portfolio_pr2.py
View File

@@ -24,7 +24,7 @@ from api.app import create_app
from src.config import Config
from src.services.portfolio_import_service import PortfolioImportService
from src.services.portfolio_risk_service import PortfolioRiskService
from src.services.portfolio_service import PortfolioService
from src.services.portfolio_service import PortfolioBusyError, PortfolioService
from src.storage import DatabaseManager
@@ -275,6 +275,40 @@ class PortfolioPr2TestCase(unittest.TestCase):
self.assertEqual(result["failed_count"], 1)
self.assertEqual(len(result["errors"]), 1)
def test_import_busy_counts_failed_not_duplicate(self) -> None:
account = self.service.create_account(name="Main", broker="Demo", market="cn", base_currency="CNY")
aid = account["id"]
with patch.object(
self.import_service.portfolio_service,
"record_trade",
side_effect=PortfolioBusyError("Portfolio ledger is busy; please retry shortly."),
):
result = self.import_service.commit_trade_records(
account_id=aid,
broker="huatai",
records=[
{
"trade_date": "2026-01-02",
"symbol": "600519",
"side": "buy",
"quantity": 10,
"price": 90,
"fee": 0.0,
"tax": 0.0,
"trade_uid": "HT-BUSY-001",
"dedup_hash": "busy-hash-001",
"market": "cn",
"currency": "CNY",
}
],
)
self.assertEqual(result["inserted_count"], 0)
self.assertEqual(result["duplicate_count"], 0)
self.assertEqual(result["failed_count"], 1)
self.assertIn("portfolio_busy", result["errors"][0])
def test_risk_threshold_boundary(self) -> None:
account = self.service.create_account(name="Main", broker="Demo", market="cn", base_currency="CNY")
aid = account["id"]

128
tests/test_portfolio_service.py
View File

@@ -4,15 +4,20 @@
from __future__ import annotations
import os
import sqlite3
import tempfile
import threading
import unittest
from datetime import date
from pathlib import Path
from unittest.mock import patch
import pandas as pd
from sqlalchemy.exc import OperationalError
from sqlalchemy import select
from src.config import Config
from src.repositories.portfolio_repo import PortfolioBusyError, PortfolioRepository
from src.services.portfolio_service import PortfolioConflictError, PortfolioOversellError, PortfolioService
from src.storage import DatabaseManager, PortfolioDailySnapshot, PortfolioPosition, PortfolioPositionLot, PortfolioTrade
@@ -453,6 +458,129 @@ class PortfolioServiceTestCase(unittest.TestCase):
self.assertEqual(len(snapshot_rows), 0)
self.assertEqual(len(lot_rows), 0)
def test_concurrent_sell_race_allows_only_one_write(self) -> None:
account = self.service.create_account(name="Main", broker="Demo", market="cn", base_currency="CNY")
aid = account["id"]
self.service.record_trade(
account_id=aid,
symbol="600519",
trade_date=date(2026, 1, 1),
side="buy",
quantity=10,
price=10,
market="cn",
currency="CNY",
)
barrier = threading.Barrier(3)
results: list[str] = []
errors: list[Exception] = []
def _worker(uid: str) -> None:
svc = PortfolioService()
barrier.wait()
try:
svc.record_trade(
account_id=aid,
symbol="600519",
trade_date=date(2026, 1, 2),
side="sell",
quantity=10,
price=11,
market="cn",
currency="CNY",
trade_uid=uid,
)
results.append(uid)
except Exception as exc: # pragma: no cover - asserted below
errors.append(exc)
threads = [
threading.Thread(target=_worker, args=(f"sell-race-{idx}",), daemon=True)
for idx in range(2)
]
for thread in threads:
thread.start()
barrier.wait()
for thread in threads:
thread.join()
self.assertEqual(len(results), 1)
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0], PortfolioOversellError)
trades = self.service.list_trade_events(account_id=aid, page=1, page_size=20)
sell_count = sum(1 for item in trades["items"] if item["side"] == "sell")
self.assertEqual(sell_count, 1)
def test_concurrent_duplicate_full_close_sell_keeps_conflict_semantics(self) -> None:
account = self.service.create_account(name="Main", broker="Demo", market="cn", base_currency="CNY")
aid = account["id"]
self.service.record_trade(
account_id=aid,
symbol="600519",
trade_date=date(2026, 1, 1),
side="buy",
quantity=10,
price=10,
market="cn",
currency="CNY",
)
barrier = threading.Barrier(3)
results: list[str] = []
errors: list[Exception] = []
def _worker() -> None:
svc = PortfolioService()
barrier.wait()
try:
svc.record_trade(
account_id=aid,
symbol="600519",
trade_date=date(2026, 1, 2),
side="sell",
quantity=10,
price=11,
market="cn",
currency="CNY",
trade_uid="dup-race-sell-1",
)
results.append("ok")
except Exception as exc: # pragma: no cover - asserted below
errors.append(exc)
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(2)]
for thread in threads:
thread.start()
barrier.wait()
for thread in threads:
thread.join()
self.assertEqual(len(results), 1)
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0], PortfolioConflictError)
self.assertIn("Duplicate trade_uid", str(errors[0]))
def test_portfolio_write_session_maps_sqlite_locked_error(self) -> None:
repo = PortfolioRepository(db_manager=self.db)
session = self.db.get_session()
stmt_exc = OperationalError(
"BEGIN IMMEDIATE",
None,
sqlite3.OperationalError("database is locked"),
)
with patch.object(self.db, "get_session", return_value=session):
with patch.object(
session.connection(),
"exec_driver_sql",
side_effect=stmt_exc,
):
with self.assertRaises(PortfolioBusyError):
with repo.portfolio_write_session():
pass
if __name__ == "__main__":
unittest.main()