feat: add P6 portfolio and watchlist alerts (#1379)

This commit is contained in:
Alfred
2026-05-21 19:15:24 +08:00
committed by GitHub
parent b7d449783c
commit 2caa1292e9
21 changed files with 2116 additions and 104 deletions
+19 -1
View File
@@ -8,9 +8,10 @@ from typing import Any, Dict, List, Literal, Optional
from pydantic import BaseModel, Field
TargetScopeValue = Literal["single_symbol"]
TargetScopeValue = Literal["single_symbol", "watchlist", "portfolio_holdings", "portfolio_account"]
SeverityValue = Literal["info", "warning", "critical"]
DryRunStatusValue = Literal["triggered", "not_triggered", "evaluation_error"]
TargetRecordStatusValue = Literal["triggered", "skipped", "degraded", "failed"]
class AlertRuleCreateRequest(BaseModel):
@@ -67,12 +68,29 @@ class AlertDeleteResponse(BaseModel):
deleted: int
class AlertRuleTargetResult(BaseModel):
target: str
display_target: Optional[str] = None
status: DryRunStatusValue
record_status: Optional[TargetRecordStatusValue] = None
triggered: bool
observed_value: Optional[Any] = None
threshold: Optional[Any] = None
message: str
class AlertRuleTestResponse(BaseModel):
rule_id: int
target_scope: Optional[str] = None
status: DryRunStatusValue
triggered: bool
observed_value: Optional[Any] = None
message: str
evaluated_count: int = 0
triggered_count: int = 0
degraded_count: int = 0
skipped_count: int = 0
target_results: List[AlertRuleTargetResult] = Field(default_factory=list)
class AlertTriggerItem(BaseModel):
@@ -162,6 +162,74 @@ describe('alertsApi', () => {
});
});
it('creates portfolio alert rules and maps batch dry-run fields', async () => {
post
.mockResolvedValueOnce({
data: {
id: 5,
name: 'portfolio stop loss',
target_scope: 'portfolio_account',
target: 'all',
alert_type: 'portfolio_stop_loss',
parameters: { mode: 'breach' },
severity: 'critical',
enabled: true,
source: 'api',
},
})
.mockResolvedValueOnce({
data: {
rule_id: 5,
target_scope: 'watchlist',
status: 'triggered',
triggered: true,
observed_value: 11,
message: 'Evaluated 2 targets',
evaluated_count: 2,
triggered_count: 1,
degraded_count: 1,
skipped_count: 0,
target_results: [
{
target: '600519',
display_target: '自选股 - 600519',
status: 'triggered',
record_status: 'triggered',
triggered: true,
observed_value: 11,
threshold: 10,
message: 'triggered',
},
],
},
});
const created = await alertsApi.createRule({
name: 'portfolio stop loss',
targetScope: 'portfolio_account',
target: 'all',
alertType: 'portfolio_stop_loss',
parameters: { mode: 'breach' },
severity: 'critical',
enabled: true,
});
const dryRun = await alertsApi.testRule(5);
expect(post).toHaveBeenNthCalledWith(1, '/api/v1/alerts/rules', {
name: 'portfolio stop loss',
target_scope: 'portfolio_account',
target: 'all',
alert_type: 'portfolio_stop_loss',
parameters: { mode: 'breach' },
severity: 'critical',
enabled: true,
});
expect(created.parameters.mode).toBe('breach');
expect(dryRun.evaluatedCount).toBe(2);
expect(dryRun.degradedCount).toBe(1);
expect(dryRun.targetResults?.[0].displayTarget).toBe('自选股 - 600519');
});
it('deletes, toggles, tests, and lists history endpoints', async () => {
deleteRequest.mockResolvedValueOnce({ data: { deleted: 1 } });
post
+1
View File
@@ -41,6 +41,7 @@ function toSnakeRulePayload(payload: AlertRuleCreateRequest): Record<string, unk
signal_period: payload.parameters.signalPeriod,
k_period: payload.parameters.kPeriod,
d_period: payload.parameters.dPeriod,
mode: payload.parameters.mode,
});
}
return request;
@@ -1,10 +1,18 @@
import type React from 'react';
import { useState } from 'react';
import { Button, Card, Checkbox, Input, Select } from '../common';
import type { AlertRuleCreateRequest, AlertSeverity, AlertType } from '../../types/alerts';
import { useEffect, useMemo, useState } from 'react';
import { portfolioApi } from '../../api/portfolio';
import type {
AlertRuleCreateRequest,
AlertSeverity,
AlertTargetScope,
AlertType,
PortfolioStopLossMode,
} from '../../types/alerts';
import type { PortfolioAccountItem } from '../../types/portfolio';
import { validateStockCode } from '../../utils/validation';
import { Button, Card, Checkbox, Input, Select } from '../common';
const ALERT_TYPE_OPTIONS = [
const SYMBOL_ALERT_TYPE_OPTIONS = [
{ value: 'price_cross', label: '价格突破' },
{ value: 'price_change_percent', label: '涨跌幅' },
{ value: 'volume_spike', label: '成交量放大' },
@@ -15,6 +23,20 @@ const ALERT_TYPE_OPTIONS = [
{ value: 'cci_threshold', label: 'CCI 阈值' },
];
const PORTFOLIO_ALERT_TYPE_OPTIONS = [
{ value: 'portfolio_stop_loss', label: '组合止损' },
{ value: 'portfolio_concentration', label: '组合集中度' },
{ value: 'portfolio_drawdown', label: '组合回撤' },
{ value: 'portfolio_price_stale', label: '组合价格状态' },
];
const TARGET_SCOPE_OPTIONS = [
{ value: 'single_symbol', label: '单标的' },
{ value: 'watchlist', label: '自选股' },
{ value: 'portfolio_holdings', label: '持仓标的' },
{ value: 'portfolio_account', label: '持仓账户' },
];
const SEVERITY_OPTIONS = [
{ value: 'info', label: '提示' },
{ value: 'warning', label: '警告' },
@@ -41,6 +63,11 @@ const CROSS_DIRECTION_OPTIONS = [
{ value: 'bearish_cross', label: '死叉' },
];
const STOP_LOSS_MODE_OPTIONS = [
{ value: 'near', label: '接近止损' },
{ value: 'breach', label: '已触发止损' },
];
const MAX_REQUESTED_DAYS = 365;
interface AlertRuleFormProps {
@@ -48,9 +75,25 @@ interface AlertRuleFormProps {
isSubmitting?: boolean;
}
function isPortfolioScope(scope: AlertTargetScope): boolean {
return scope === 'portfolio_holdings' || scope === 'portfolio_account';
}
function defaultAlertTypeForScope(scope: AlertTargetScope): AlertType {
return scope === 'portfolio_account' ? 'portfolio_stop_loss' : 'price_cross';
}
function optionsForScope(scope: AlertTargetScope) {
return scope === 'portfolio_account' ? PORTFOLIO_ALERT_TYPE_OPTIONS : SYMBOL_ALERT_TYPE_OPTIONS;
}
export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmitting = false }) => {
const [name, setName] = useState('');
const [targetScope, setTargetScope] = useState<AlertTargetScope>('single_symbol');
const [target, setTarget] = useState('');
const [portfolioTarget, setPortfolioTarget] = useState('all');
const [accounts, setAccounts] = useState<PortfolioAccountItem[]>([]);
const [accountsError, setAccountsError] = useState<string | null>(null);
const [alertType, setAlertType] = useState<AlertType>('price_cross');
const [severity, setSeverity] = useState<AlertSeverity>('warning');
const [enabled, setEnabled] = useState(true);
@@ -58,6 +101,7 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
const [changeDirection, setChangeDirection] = useState<'up' | 'down'>('up');
const [thresholdDirection, setThresholdDirection] = useState<'above' | 'below'>('above');
const [crossDirection, setCrossDirection] = useState<'bullish_cross' | 'bearish_cross'>('bullish_cross');
const [stopLossMode, setStopLossMode] = useState<PortfolioStopLossMode>('near');
const [price, setPrice] = useState('');
const [changePct, setChangePct] = useState('');
const [multiplier, setMultiplier] = useState('');
@@ -71,6 +115,34 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
const [dPeriod, setDPeriod] = useState('3');
const [formError, setFormError] = useState<string | null>(null);
useEffect(() => {
if (!isPortfolioScope(targetScope)) return undefined;
let cancelled = false;
void portfolioApi.getAccounts(false)
.then((response) => {
if (cancelled) return;
setAccounts(response.accounts ?? []);
setAccountsError(null);
})
.catch((error: unknown) => {
if (cancelled) return;
setAccounts([]);
setAccountsError(error instanceof Error ? error.message : '账户加载失败');
});
return () => {
cancelled = true;
};
}, [targetScope]);
const alertTypeOptions = useMemo(() => optionsForScope(targetScope), [targetScope]);
const portfolioTargetOptions = useMemo(() => [
{ value: 'all', label: '全部账户' },
...accounts.map((account) => ({
value: String(account.id),
label: `${account.name} #${account.id}`,
})),
], [accounts]);
const resetParameters = (nextType: AlertType) => {
if (nextType === 'price_cross') {
setPriceDirection('above');
@@ -101,6 +173,8 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
setThresholdDirection('above');
setPeriod('14');
setThreshold('');
} else if (nextType === 'portfolio_stop_loss') {
setStopLossMode('near');
}
};
@@ -153,71 +227,104 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
return true;
};
const handleSubmit = async (event: React.FormEvent<HTMLFormElement>) => {
event.preventDefault();
const targetValidation = validateStockCode(target);
if (!targetValidation.valid) {
setFormError(targetValidation.message ?? '股票代码格式不正确');
return;
}
let parameters: AlertRuleCreateRequest['parameters'];
const buildParameters = (): AlertRuleCreateRequest['parameters'] | null => {
if (alertType === 'price_cross') {
const parsedPrice = parsePositiveNumber(price, '价格阈值');
if (parsedPrice == null) return;
parameters = { direction: priceDirection, price: parsedPrice };
} else if (alertType === 'price_change_percent') {
if (parsedPrice == null) return null;
return { direction: priceDirection, price: parsedPrice };
}
if (alertType === 'price_change_percent') {
const parsedChangePct = parsePositiveNumber(changePct, '涨跌幅阈值');
if (parsedChangePct == null) return;
parameters = { direction: changeDirection, changePct: parsedChangePct };
} else if (alertType === 'volume_spike') {
if (parsedChangePct == null) return null;
return { direction: changeDirection, changePct: parsedChangePct };
}
if (alertType === 'volume_spike') {
const parsedMultiplier = parsePositiveNumber(multiplier, '成交量倍数');
if (parsedMultiplier == null) return;
parameters = { multiplier: parsedMultiplier };
} else if (alertType === 'ma_price_cross') {
if (parsedMultiplier == null) return null;
return { multiplier: parsedMultiplier };
}
if (alertType === 'ma_price_cross') {
const parsedWindow = parseIntegerInRange(window, '均线周期');
if (parsedWindow == null) return;
parameters = { direction: thresholdDirection, window: parsedWindow };
} else if (alertType === 'rsi_threshold') {
if (parsedWindow == null) return null;
return { direction: thresholdDirection, window: parsedWindow };
}
if (alertType === 'rsi_threshold') {
const parsedPeriod = parseIntegerInRange(period, 'RSI 周期');
const parsedThreshold = parseRsiThreshold(threshold);
if (parsedPeriod == null || parsedThreshold == null) return;
parameters = { direction: thresholdDirection, period: parsedPeriod, threshold: parsedThreshold };
} else if (alertType === 'macd_cross') {
if (parsedPeriod == null || parsedThreshold == null) return null;
return { direction: thresholdDirection, period: parsedPeriod, threshold: parsedThreshold };
}
if (alertType === 'macd_cross') {
const parsedFast = parseIntegerInRange(fastPeriod, '快线周期');
const parsedSlow = parseIntegerInRange(slowPeriod, '慢线周期');
const parsedSignal = parseIntegerInRange(signalPeriod, '信号周期');
if (parsedFast == null || parsedSlow == null || parsedSignal == null) return;
if (parsedFast == null || parsedSlow == null || parsedSignal == null) return null;
if (parsedFast >= parsedSlow) {
setFormError('快线周期必须小于慢线周期');
return;
return null;
}
if (!ensureRequiredBarsWithinLimit('MACD', parsedSlow + parsedSignal + 1)) return;
parameters = {
if (!ensureRequiredBarsWithinLimit('MACD', parsedSlow + parsedSignal + 1)) return null;
return {
direction: crossDirection,
fastPeriod: parsedFast,
slowPeriod: parsedSlow,
signalPeriod: parsedSignal,
};
} else if (alertType === 'kdj_cross') {
}
if (alertType === 'kdj_cross') {
const parsedPeriod = parseIntegerInRange(period, 'KDJ 周期');
const parsedK = parseIntegerInRange(kPeriod, 'K 平滑周期');
const parsedD = parseIntegerInRange(dPeriod, 'D 平滑周期');
if (parsedPeriod == null || parsedK == null || parsedD == null) return;
if (!ensureRequiredBarsWithinLimit('KDJ', parsedPeriod + parsedK + parsedD + 1)) return;
parameters = { direction: crossDirection, period: parsedPeriod, kPeriod: parsedK, dPeriod: parsedD };
} else {
if (parsedPeriod == null || parsedK == null || parsedD == null) return null;
if (!ensureRequiredBarsWithinLimit('KDJ', parsedPeriod + parsedK + parsedD + 1)) return null;
return { direction: crossDirection, period: parsedPeriod, kPeriod: parsedK, dPeriod: parsedD };
}
if (alertType === 'cci_threshold') {
const parsedPeriod = parseIntegerInRange(period, 'CCI 周期');
const parsedThreshold = parseFiniteNumber(threshold, 'CCI 阈值');
if (parsedPeriod == null || parsedThreshold == null) return;
parameters = { direction: thresholdDirection, period: parsedPeriod, threshold: parsedThreshold };
if (parsedPeriod == null || parsedThreshold == null) return null;
return { direction: thresholdDirection, period: parsedPeriod, threshold: parsedThreshold };
}
if (alertType === 'portfolio_stop_loss') {
return { mode: stopLossMode };
}
return {};
};
const handleScopeChange = (value: string) => {
const nextScope = value as AlertTargetScope;
const nextType = defaultAlertTypeForScope(nextScope);
setTargetScope(nextScope);
setAlertType(nextType);
setPortfolioTarget('all');
resetParameters(nextType);
setFormError(null);
};
const handleSubmit = async (event: React.FormEvent<HTMLFormElement>) => {
event.preventDefault();
let resolvedTarget = target.trim();
if (targetScope === 'single_symbol') {
const targetValidation = validateStockCode(target);
if (!targetValidation.valid) {
setFormError(targetValidation.message ?? '股票代码格式不正确');
return;
}
resolvedTarget = targetValidation.normalized;
} else if (targetScope === 'watchlist') {
resolvedTarget = 'default';
} else {
resolvedTarget = portfolioTarget;
}
const parameters = buildParameters();
if (parameters == null) return;
setFormError(null);
const submitted = await onSubmit({
name: name.trim() || undefined,
targetScope: 'single_symbol',
target: targetValidation.normalized,
targetScope,
target: resolvedTarget,
alertType,
parameters,
severity,
@@ -226,6 +333,7 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
if (submitted === false) return;
setName('');
setTarget('');
setPortfolioTarget('all');
setPrice('');
setChangePct('');
setMultiplier('');
@@ -241,6 +349,42 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
setEnabled(true);
};
const renderTargetControl = () => {
if (targetScope === 'single_symbol') {
return (
<Input
label="标的代码"
value={target}
onChange={(event) => setTarget(event.target.value)}
placeholder="600519 / AAPL / hk00700"
disabled={isSubmitting}
/>
);
}
if (targetScope === 'watchlist') {
return (
<Input
label="目标"
value="default"
onChange={() => undefined}
disabled
/>
);
}
return (
<div className="space-y-2">
<Select
label="账户"
value={portfolioTarget}
options={portfolioTargetOptions}
disabled={isSubmitting}
onChange={setPortfolioTarget}
/>
{accountsError ? <p role="alert" className="text-xs text-warning">{accountsError}</p> : null}
</div>
);
};
return (
<Card title="创建告警规则" subtitle="Web 告警中心" variant="bordered" padding="md">
<form className="space-y-4" noValidate onSubmit={(event) => void handleSubmit(event)}>
@@ -252,17 +396,18 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
placeholder="可选,例如 茅台价格突破"
disabled={isSubmitting}
/>
<Input
label="标的代码"
value={target}
onChange={(event) => setTarget(event.target.value)}
placeholder="600519 / AAPL / hk00700"
<Select
label="目标范围"
value={targetScope}
options={TARGET_SCOPE_OPTIONS}
disabled={isSubmitting}
onChange={handleScopeChange}
/>
{renderTargetControl()}
<Select
label="规则类型"
value={alertType}
options={ALERT_TYPE_OPTIONS}
options={alertTypeOptions}
disabled={isSubmitting}
onChange={(value) => {
const nextType = value as AlertType;
@@ -501,6 +646,16 @@ export const AlertRuleForm: React.FC<AlertRuleFormProps> = ({ onSubmit, isSubmit
</div>
) : null}
{alertType === 'portfolio_stop_loss' ? (
<Select
label="止损模式"
value={stopLossMode}
options={STOP_LOSS_MODE_OPTIONS}
disabled={isSubmitting}
onChange={(value) => setStopLossMode(value as PortfolioStopLossMode)}
/>
) : null}
<div className="flex flex-col gap-3 sm:flex-row sm:items-center sm:justify-between">
<Checkbox
label="创建后立即启用"
@@ -30,6 +30,10 @@ const ALERT_TYPE_FILTER_OPTIONS = [
{ value: 'macd_cross', label: 'MACD 金叉/死叉' },
{ value: 'kdj_cross', label: 'KDJ 金叉/死叉' },
{ value: 'cci_threshold', label: 'CCI 阈值' },
{ value: 'portfolio_stop_loss', label: '组合止损' },
{ value: 'portfolio_concentration', label: '组合集中度' },
{ value: 'portfolio_drawdown', label: '组合回撤' },
{ value: 'portfolio_price_stale', label: '组合价格状态' },
];
const typeLabel: Record<AlertType, string> = {
@@ -41,6 +45,10 @@ const typeLabel: Record<AlertType, string> = {
macd_cross: 'MACD 金叉/死叉',
kdj_cross: 'KDJ 金叉/死叉',
cci_threshold: 'CCI 阈值',
portfolio_stop_loss: '组合止损',
portfolio_concentration: '组合集中度',
portfolio_drawdown: '组合回撤',
portfolio_price_stale: '组合价格状态',
};
const severityLabel: Record<string, string> = {
@@ -49,6 +57,13 @@ const severityLabel: Record<string, string> = {
critical: '严重',
};
const scopeLabel: Record<string, string> = {
single_symbol: '单标的',
watchlist: '自选股',
portfolio_holdings: '持仓标的',
portfolio_account: '持仓账户',
};
function formatParameters(rule: AlertRuleItem): string {
if (rule.alertType === 'price_cross') {
return `${rule.parameters.direction === 'below' ? '下破' : '上破'} ${rule.parameters.price ?? '--'}`;
@@ -72,6 +87,12 @@ function formatParameters(rule: AlertRuleItem): string {
}
return `KDJ(${rule.parameters.period ?? '--'},${rule.parameters.kPeriod ?? '--'},${rule.parameters.dPeriod ?? '--'}) ${direction}`;
}
if (rule.alertType === 'portfolio_stop_loss') {
return rule.parameters.mode === 'breach' ? '已触发止损' : '接近止损';
}
if (rule.alertType === 'portfolio_concentration') return 'top_weight_pct';
if (rule.alertType === 'portfolio_drawdown') return 'max_drawdown_pct';
if (rule.alertType === 'portfolio_price_stale') return 'price_stale / price_available';
return `CCI${rule.parameters.period ?? '--'} ${rule.parameters.direction === 'below' ? '下穿' : '上穿'} ${rule.parameters.threshold ?? '--'}`;
}
@@ -79,6 +100,18 @@ function isCoolingDown(rule: AlertRuleItem): boolean {
return rule.cooldownActive === true;
}
function formatTarget(rule: AlertRuleItem): string {
if (rule.targetScope === 'watchlist') return 'default';
if (rule.targetScope === 'portfolio_account' || rule.targetScope === 'portfolio_holdings') {
return rule.target === 'all' ? '全部账户' : `账户 ${rule.target}`;
}
return rule.target;
}
function hasChildTargetCooldown(rule: AlertRuleItem): boolean {
return rule.targetScope === 'watchlist' || rule.targetScope === 'portfolio_holdings';
}
interface AlertRuleListProps {
rules: AlertRuleItem[];
total: number;
@@ -172,7 +205,10 @@ export const AlertRuleList: React.FC<AlertRuleListProps> = ({
<div className="font-medium text-foreground">{rule.name}</div>
<div className="mt-1 text-xs text-muted-text">来源:{rule.source}</div>
</td>
<td className="px-3 py-3 font-mono text-secondary-text">{rule.target}</td>
<td className="px-3 py-3 text-secondary-text">
<div className="font-mono">{formatTarget(rule)}</div>
<div className="mt-1 text-xs">{scopeLabel[rule.targetScope] ?? rule.targetScope}</div>
</td>
<td className="px-3 py-3">
<div className="flex flex-col items-start gap-1">
<Badge variant="info">{typeLabel[rule.alertType]}</Badge>
@@ -190,6 +226,9 @@ export const AlertRuleList: React.FC<AlertRuleListProps> = ({
<td className="px-3 py-3 text-xs text-secondary-text">
<div>{isCoolingDown(rule) ? '冷却中' : '未冷却'}</div>
<div className="mt-1">{formatDateTime(rule.cooldownUntil)}</div>
{hasChildTargetCooldown(rule) ? (
<div className="mt-1 text-muted-text">子目标见触发历史</div>
) : null}
</td>
<td className="px-3 py-3 text-xs text-secondary-text">{formatDateTime(rule.updatedAt ?? rule.createdAt)}</td>
<td className="px-3 py-3">
@@ -2,12 +2,24 @@ import { fireEvent, render, screen, waitFor } from '@testing-library/react';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { AlertRuleForm } from '../AlertRuleForm';
const { getAccounts } = vi.hoisted(() => ({
getAccounts: vi.fn(),
}));
vi.mock('../../../api/portfolio', () => ({
portfolioApi: {
getAccounts,
},
}));
describe('AlertRuleForm', () => {
const onSubmit = vi.fn();
beforeEach(() => {
onSubmit.mockReset();
onSubmit.mockResolvedValue(undefined);
getAccounts.mockReset();
getAccounts.mockResolvedValue({ accounts: [{ id: 9, name: 'Main', market: 'us', baseCurrency: 'USD', isActive: true }] });
});
it('submits a price_cross rule payload', async () => {
@@ -160,6 +172,53 @@ describe('AlertRuleForm', () => {
expect(onSubmit).not.toHaveBeenCalled();
});
it('filters alert types and submits a watchlist rule payload', async () => {
render(<AlertRuleForm onSubmit={onSubmit} />);
fireEvent.change(screen.getByLabelText('目标范围'), { target: { value: 'watchlist' } });
expect(screen.queryByText('组合止损')).not.toBeInTheDocument();
fireEvent.change(screen.getByLabelText('价格阈值'), { target: { value: '10' } });
fireEvent.click(screen.getByRole('button', { name: '创建规则' }));
await waitFor(() => {
expect(onSubmit).toHaveBeenCalledWith(expect.objectContaining({
targetScope: 'watchlist',
target: 'default',
alertType: 'price_cross',
parameters: { direction: 'above', price: 10 },
}));
});
});
it('loads accounts and submits portfolio stop-loss mode', async () => {
render(<AlertRuleForm onSubmit={onSubmit} />);
fireEvent.change(screen.getByLabelText('目标范围'), { target: { value: 'portfolio_account' } });
await waitFor(() => expect(getAccounts).toHaveBeenCalledWith(false));
expect(screen.queryByText('价格突破')).not.toBeInTheDocument();
fireEvent.change(screen.getByLabelText('账户'), { target: { value: '9' } });
fireEvent.change(screen.getByLabelText('止损模式'), { target: { value: 'breach' } });
fireEvent.click(screen.getByRole('button', { name: '创建规则' }));
await waitFor(() => {
expect(onSubmit).toHaveBeenCalledWith(expect.objectContaining({
targetScope: 'portfolio_account',
target: '9',
alertType: 'portfolio_stop_loss',
parameters: { mode: 'breach' },
}));
});
});
it('keeps all account option when account loading fails', async () => {
getAccounts.mockRejectedValueOnce(new Error('boom'));
render(<AlertRuleForm onSubmit={onSubmit} />);
fireEvent.change(screen.getByLabelText('目标范围'), { target: { value: 'portfolio_holdings' } });
expect(await screen.findByRole('alert')).toHaveTextContent('boom');
expect(screen.getByLabelText('账户')).toHaveValue('all');
});
it('keeps form values when submit reports failure', async () => {
onSubmit.mockResolvedValueOnce(false);
render(<AlertRuleForm onSubmit={onSubmit} />);
@@ -117,6 +117,43 @@ describe('AlertRuleList', () => {
expect(screen.getByText('未冷却')).toBeInTheDocument();
});
it('renders portfolio scope labels and child-target cooldown hint', () => {
renderList({
rules: [
{
id: 4,
name: '持仓 RSI',
targetScope: 'portfolio_holdings',
target: 'all',
alertType: 'rsi_threshold',
parameters: { direction: 'below', period: 12, threshold: 30 },
severity: 'warning',
enabled: true,
source: 'api',
cooldownActive: false,
},
{
id: 5,
name: '组合止损',
targetScope: 'portfolio_account',
target: '9',
alertType: 'portfolio_stop_loss',
parameters: { mode: 'breach' },
severity: 'critical',
enabled: true,
source: 'api',
cooldownActive: false,
},
],
});
expect(screen.getByText('持仓标的')).toBeInTheDocument();
expect(screen.getByText('子目标见触发历史')).toBeInTheDocument();
expect(screen.getByText('账户 9')).toBeInTheDocument();
expect(screen.getAllByText('组合止损').length).toBeGreaterThan(0);
expect(screen.getByText('已触发止损')).toBeInTheDocument();
});
it('runs test and toggles enabled state', () => {
renderList();
+36 -11
View File
@@ -40,6 +40,41 @@ function testVariant(result: AlertRuleTestResponse): 'success' | 'warning' | 'da
return result.triggered ? 'success' : 'warning';
}
function renderTestResultMessage(result: AlertRuleTestResponse): React.ReactNode {
const targetResults = result.targetResults ?? [];
return (
<div className="space-y-2">
<div>
{result.message}
{' · 状态:'}
{result.status}
{' · 触发:'}
{result.triggered ? '是' : '否'}
{' · 观察值:'}
{result.observedValue == null ? '--' : String(result.observedValue)}
</div>
{result.evaluatedCount != null && result.evaluatedCount > 1 ? (
<div className="text-xs">
评估 {result.evaluatedCount} · 触发 {result.triggeredCount ?? 0} · 降级 {result.degradedCount ?? 0} · 跳过 {result.skippedCount ?? 0}
</div>
) : null}
{targetResults.length > 1 ? (
<div className="grid gap-1 text-xs">
{targetResults.slice(0, 20).map((item) => (
<div key={`${item.target}-${item.status}`} className="flex flex-wrap justify-between gap-2">
<span>{item.displayTarget ?? item.target}</span>
<span>
{item.status}
{item.recordStatus ? ` / ${item.recordStatus}` : ''}
</span>
</div>
))}
</div>
) : null}
</div>
);
}
const notificationChannelLabel: Record<string, string> = {
__cooldown__: '业务冷却',
__cooldown_read_failed__: '冷却读取失败',
@@ -276,17 +311,7 @@ const AlertsPage: React.FC = () => {
<InlineAlert
title="测试结果"
variant={testVariant(testResult)}
message={(
<span>
{testResult.message}
{' · 状态:'}
{testResult.status}
{' · 触发:'}
{testResult.triggered ? '是' : '否'}
{' · 观察值:'}
{testResult.observedValue == null ? '--' : String(testResult.observedValue)}
</span>
)}
message={renderTestResultMessage(testResult)}
/>
) : null}
</div>
@@ -35,6 +35,12 @@ vi.mock('../../api/alerts', () => ({
},
}));
vi.mock('../../api/portfolio', () => ({
portfolioApi: {
getAccounts: vi.fn().mockResolvedValue({ accounts: [] }),
},
}));
const parsedError = {
title: '加载失败',
message: '告警 API 不可用',
@@ -131,6 +137,48 @@ describe('AlertsPage', () => {
expect(screen.queryByText(/realtime_quote/)).not.toBeInTheDocument();
});
it('renders batch dry-run summary and target results', async () => {
testRule.mockResolvedValueOnce({
ruleId: 1,
targetScope: 'watchlist',
status: 'triggered',
triggered: true,
observedValue: 11,
message: 'Evaluated 2 targets',
evaluatedCount: 2,
triggeredCount: 1,
degradedCount: 1,
skippedCount: 0,
targetResults: [
{
target: '600519',
displayTarget: '自选股 - 600519',
status: 'triggered',
recordStatus: 'triggered',
triggered: true,
observedValue: 11,
message: 'triggered',
},
{
target: '000001',
displayTarget: '自选股 - 000001',
status: 'not_triggered',
recordStatus: 'degraded',
triggered: false,
observedValue: null,
message: 'degraded',
},
],
});
render(<AlertsPage />);
fireEvent.click(await screen.findByRole('button', { name: '测试' }));
expect(await screen.findByText(/评估 2 · 触发 1 · 降级 1 · 跳过 0/)).toBeInTheDocument();
expect(screen.getByText('自选股 - 600519')).toBeInTheDocument();
expect(screen.getByText(/not_triggered \/ degraded/)).toBeInTheDocument();
});
it('creates a rule through the page form and reloads rules', async () => {
render(<AlertsPage />);
+25 -2
View File
@@ -6,10 +6,15 @@ export type AlertType =
| 'rsi_threshold'
| 'macd_cross'
| 'kdj_cross'
| 'cci_threshold';
| 'cci_threshold'
| 'portfolio_stop_loss'
| 'portfolio_concentration'
| 'portfolio_drawdown'
| 'portfolio_price_stale';
export type AlertSeverity = 'info' | 'warning' | 'critical';
export type AlertTargetScope = 'single_symbol';
export type AlertTargetScope = 'single_symbol' | 'watchlist' | 'portfolio_holdings' | 'portfolio_account';
export type AlertDirection = 'above' | 'below' | 'up' | 'down' | 'bullish_cross' | 'bearish_cross';
export type PortfolioStopLossMode = 'near' | 'breach';
export type AlertDryRunStatus = 'triggered' | 'not_triggered' | 'evaluation_error';
export type AlertTriggerStatus = 'triggered' | 'skipped' | 'degraded' | 'failed';
@@ -26,6 +31,7 @@ export interface AlertRuleParameters {
signalPeriod?: number;
kPeriod?: number;
dPeriod?: number;
mode?: PortfolioStopLossMode;
}
export interface AlertRuleItem {
@@ -70,10 +76,27 @@ export interface AlertDeleteResponse {
export interface AlertRuleTestResponse {
ruleId: number;
targetScope?: AlertTargetScope | string | null;
status: AlertDryRunStatus;
triggered: boolean;
observedValue?: unknown;
message: string;
evaluatedCount?: number;
triggeredCount?: number;
degradedCount?: number;
skippedCount?: number;
targetResults?: AlertRuleTargetResult[];
}
export interface AlertRuleTargetResult {
target: string;
displayTarget?: string | null;
status: AlertDryRunStatus;
recordStatus?: AlertTriggerStatus | null;
triggered: boolean;
observedValue?: unknown;
threshold?: unknown;
message: string;
}
export interface AlertTriggerItem {
+1
View File
@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
- [新功能] 告警中心 P4 记录真实通知尝试结果,并为持久化规则新增可查询的业务冷却状态。
- [修复] 持仓快照在当天刷新时优先使用实时行情重算当前价、市值与未实现盈亏,避免复用旧收盘价导致页面刷新后盈亏不变。
- [新功能] 告警中心 P5 支持 MA、RSI、MACD、KDJ、CCI 日线技术指标规则,并复用现有触发历史、通知结果和持久化冷却链路。
- [新功能] 告警中心 P6 支持自选股、持仓标的和持仓账户联动规则,复用现有触发历史、通知结果与业务冷却链路。
- [改进] 将 RSI 计算口径从 SMA 调整为 Wilder's EMA / SMMA,统一分析报告与告警阈值口径。
- [改进] 大盘复盘将红绿灯与盘面温度合并为终端友好的盘面信号分数,移除色块进度条与重复温度行。
- [改进] 大盘复盘近三日市场线索改为标题与来源链接列表,移除摘要片段,降低中英混排和误读风险。
+67
View File
@@ -265,6 +265,72 @@ P5 不做:
- 不支持 legacy `AGENT_EVENT_ALERT_RULES_JSON` 技术指标规则。
- 不引入 DSL、规则引擎、新数据库表或分析报告 pipeline 内的技术指标规则引擎。
## P6 持仓与自选股联动
P6 在现有 Alert API、Web 告警中心和 `src/services/alert_worker.py` 评估链路中新增 `watchlist`、`portfolio_holdings`、`portfolio_account` 三类目标范围。规则仍写入 `alert_rules`,触发、降级、失败、通知结果和持久化冷却继续复用 P2-P4 的 `alert_triggers`、`alert_notifications` 与 `alert_cooldowns` 语义,不新增表或迁移。
### P6 scope/type 矩阵
| `target_scope` | `target` | 允许的 `alert_type` | 评估方式 |
| --- | --- | --- | --- |
| `single_symbol` | 股票代码 | P1 三类价格/成交量规则 + P5 技术指标 | 单规则单标的 |
| `watchlist` | `default` | P1 三类价格/成交量规则 + P5 技术指标 | 每轮刷新并读取当前 `STOCK_LIST`,按股票代码展开 |
| `portfolio_holdings` | `all` 或 active account ID | P1 三类价格/成交量规则 + P5 技术指标 | 从持仓 snapshot 的非零持仓展开 symbol,按 symbol 去重 |
| `portfolio_account` | `all` 或 active account ID | `portfolio_stop_loss`、`portfolio_concentration`、`portfolio_drawdown`、`portfolio_price_stale` | 账户级风险评估,不展开为单标的 |
创建/更新规则时,`watchlist` / `portfolio_holdings` 不把父级 `target` 当股票代码校验;`portfolio_account` 禁止 price/volume/技术指标类型;`portfolio_holdings` 和 `portfolio_account` 在 `target=<id>` 时会校验账户存在且 active,不存在返回 HTTP 400 + `validation_error`。legacy `AGENT_EVENT_ALERT_RULES_JSON` 不支持 watchlist、portfolio 或技术指标扩展,继续仅支持 `single_symbol` 的 `price_cross`、`price_change_percent`、`volume_spike`。
### Target Identity Contract
P6 将可展示目标与可持久化目标分离:
| 场景 | `effective_target` | `display_target` |
| --- | --- | --- |
| `single_symbol` | `<symbol>` | `<symbol>` |
| `watchlist` 展开子目标 | `<symbol>` | `自选股 - <symbol>` |
| `portfolio_holdings` 展开子目标 | `<symbol>` | `持仓 - <symbol>` |
| `portfolio_account target=all` | `account:all` | `全部账户` |
| `portfolio_account target=<id>` | `account:<id>` | `账户 <id>` |
- `alert_triggers.target`、`alert_cooldowns.target`、P4 `rule_id + target + data_source + data_timestamp` 去重全部使用 `effective_target`。
- `RuntimeAlertRule.key` 对展开后的子目标使用 `{parent_key}|{effective_target}`,避免 DB cooldown 读取失败时的进程内 fallback 把同一父规则下的不同子目标互相 suppress。
- `display_target` 不写入 `alert_triggers.target`,仅用于通知标题、dry-run `target_results` 和 Web 展示。
- P6 不做跨规则同标的通知合并;同一股票若同时命中 watchlist 子规则和独立 `single_symbol` 规则,会按每条规则独立记录和通知。
### Dry-run 聚合
- `POST /api/v1/alerts/rules/{rule_id}/test` 对批量规则返回聚合字段:`evaluated_count`、`triggered_count`、`degraded_count`、`skipped_count`、`target_results`。
- 展开目标 soft cap 为 100;dry-run 中超过 soft cap 的目标记为 `degraded` 聚合结果并写日志。worker 运行时只评估前 100 个展开目标并写 warning,不为 overflow 本身写 `alert_triggers` 历史。
- dry-run 使用受限并发评估,单目标超时 10 秒,总评估超时 30 秒;未完成目标记为 `skipped`。
- 任一目标 triggered 时顶层 `status=triggered`;无触发但存在成功评估、skipped 或 degraded 时顶层 `status=not_triggered`;无法展开或全部失败时才返回 `evaluation_error`。
- 空 watchlist / 空 holdings:dry-run 返回 `not_triggered` 并在 `target_results` 中给出 `record_status=skipped`;worker 会写 `skipped` 历史。
- `degraded_count` 统计全部展开评估结果中 `record_status=degraded` 的条目;`target_results` 仅展示前 20 条,排序为 triggered 优先,其次 degraded/failed,再按 target 排序。
### 持仓风险规则
| `alert_type` | 参数 | 观察值 | 触发语义 |
| --- | --- | --- | --- |
| `portfolio_stop_loss` | `mode=near|breach`,默认 `near` | 受影响标的最大 `loss_pct` | `near` 使用 `stop_loss.near_alert`,`breach` 只统计 `is_triggered=true` 的 items;每账户每轮最多一条 trigger |
| `portfolio_concentration` | - | `concentration.top_weight_pct` | `top_weight_pct >= portfolio_risk_concentration_alert_pct` |
| `portfolio_drawdown` | - | `drawdown.max_drawdown_pct` | 复用 `PortfolioRiskService` 的 `drawdown.alert`;`current_drawdown_pct` 写 diagnostics |
| `portfolio_price_stale` | - | stale/missing 价格持仓数量 | 任一 position `price_stale=true` 或 `price_available=false` |
portfolio diagnostics 必含 `account_id`(或 `all`)、`currency`、`as_of`、`price_stale`、`fx_stale`、`data_available`、`top_affected_symbols`。`portfolio_stop_loss`、`portfolio_concentration`、`portfolio_drawdown` 复用 `PortfolioRiskService.get_risk_report()`;`portfolio_price_stale` 复用 `PortfolioService.get_portfolio_snapshot()` 的 position price metadata。
### Web 与 cooldown 摘要
- Web 创建表单新增目标范围选择;`watchlist` / `portfolio_holdings` 只显示 price/volume/P5 技术指标类型,`portfolio_account` 只显示四类 portfolio 风险类型。
- `portfolio_holdings` / `portfolio_account` 加载账户列表失败时,表单保留 `all` 选项并展示错误。
- 规则列表上的 `cooldown_active` 对 `single_symbol` 和 `portfolio_account` 准确;`watchlist` / `portfolio_holdings` 是父规则摘要,不代表每个子目标的冷却状态,子目标冷却以触发历史和 `effective_target` 为准。
- dry-run UI 展示聚合计数和最多 20 条 `target_results` 明细。
P6 不做:
- 不做 P7 Market Light。
- 不做财报日前、分红除权日前提醒;这类规则需要稳定日期契约后另起 follow-up。
- 不做 sector 级集中度告警;P6 集中度使用 symbol 维度 `top_weight_pct`。
- 不做跨规则同标的通知合并、分钟线、多市场时区精确判定或 legacy JSON 扩展。
## Phase 边界
- P0:本文档、契约、存储评估和兼容测试。
@@ -294,3 +360,4 @@ P5 不做:
- P3 是 Web 和文档改动。最小回滚方式是 revert P3 PR;不会删除已有规则、触发历史或 legacy JSON 配置。
- P4 新增 `alert_cooldowns` SQLite 表并开始写入 `alert_notifications`。最小回滚方式是 revert P4 PR;已经创建的 `alert_cooldowns`、`alert_triggers`、`alert_notifications` 数据不会自动删除。如需清理,需要维护者确认后手动删除对应表或记录。
- P5 新增 Alert API/Web 支持的技术指标规则。最小回滚方式是 revert P5 PR;已创建的 P5 `alert_rules` 记录不会自动删除,旧代码会在 worker 加载阶段 skip unsupported `alert_type`,不影响 legacy 三类规则执行。如需清理,需要维护者确认后手动删除相关规则记录。
- P6 新增 Alert API/Web 支持的 watchlist、portfolio holdings 与 portfolio account 规则。最小回滚方式是 revert P6 PR;没有新表或迁移,已创建的 P6 `alert_rules` 会保留。回滚前建议 disable/delete 非 `single_symbol` 的 P6 规则;否则旧 worker 可能把 `watchlist` / `portfolio_holdings` 的父级 `target` 当作股票代码评估并产生 failed/skipped 噪声,portfolio 专用 `alert_type` 会在 worker 加载阶段被 skip。
+6 -2
View File
@@ -1316,7 +1316,7 @@ A: 检查是否启用了 Actions,以及 cron 表达式是否正确(注意是
## Agent 事件告警监控
`AGENT_EVENT_MONITOR_ENABLED=true` 后,schedule 模式会按 `AGENT_EVENT_MONITOR_INTERVAL_MINUTES` 运行告警 worker。worker 每轮读取 Alert API 创建并启用的持久化规则,同时继续兼容 `AGENT_EVENT_ALERT_RULES_JSON` 中的 legacy 规则;触发后仍发送到现有通知渠道。Alert API / Web 持久化规则支持实时价、涨跌幅、成交量和日线技术指标;legacy JSON 仍仅支持三类基础规则。
`AGENT_EVENT_MONITOR_ENABLED=true` 后,schedule 模式会按 `AGENT_EVENT_MONITOR_INTERVAL_MINUTES` 运行告警 worker。worker 每轮读取 Alert API 创建并启用的持久化规则,同时继续兼容 `AGENT_EVENT_ALERT_RULES_JSON` 中的 legacy 规则;触发后仍发送到现有通知渠道。Alert API / Web 持久化规则支持实时价、涨跌幅、成交量、日线技术指标,以及 `watchlist`、`portfolio_holdings`、`portfolio_account` 三类联动目标;legacy JSON 仍仅支持三类基础规则。
> 兼容与迁移说明:本节记录当前事件告警规则(含 `price_change_percent`)运行时行为,未变更模型名、provider、Base URL、LiteLLM、`OPENAI_*`、`DEEPSEEK_*`、`GEMINI_*` 等外部模型/API 配置语义。legacy JSON 不会被自动迁移、删除或改写;若需回退,删除或关闭 `AGENT_EVENT_MONITOR_ENABLED` 即可停止后台告警 worker。
@@ -1330,6 +1330,10 @@ A: 检查是否启用了 Actions,以及 cron 表达式是否正确(注意是
| `macd_cross` | `bullish_cross` / `bearish_cross` | `fast_period`、`slow_period`、`signal_period` | DIF/DEA 边缘金叉或死叉 |
| `kdj_cross` | `bullish_cross` / `bearish_cross` | `period`、`k_period`、`d_period` | K/D 边缘金叉或死叉 |
| `cci_threshold` | `above` / `below` | `period`、`threshold` | CCI 边缘上穿或下穿阈值 |
| `portfolio_stop_loss` | `mode=near|breach` | - | 账户级止损接近或触发 |
| `portfolio_concentration` | - | - | 账户级 symbol 集中度 |
| `portfolio_drawdown` | - | - | 账户级最大回撤告警 |
| `portfolio_price_stale` | - | - | 持仓价格 stale 或 missing |
示例:
@@ -1341,7 +1345,7 @@ AGENT_EVENT_ALERT_RULES_JSON=[{"stock_code":"600519","alert_type":"price_cross",
worker 会把 `triggered`、`skipped`、`degraded`、`failed` 写入 `alert_triggers` 作为评估历史;正常未触发不写历史。DB 持久化规则的 `triggered` 历史按 `rule_id + target + data_source + data_timestamp` 对同一数据点做 best-effort 去重,重复命中会复用最早一条触发记录,`data_timestamp` 缺失时不去重。真实触发后会把每个通知渠道的 attempt 写入 `alert_notifications`,并为 Alert API 创建的持久化规则写入 `alert_cooldowns` 业务冷却状态;若读取持久化冷却失败,worker 会临时使用进程内 fingerprint 防止 DB 异常期间重复推送。legacy `AGENT_EVENT_ALERT_RULES_JSON` 规则继续使用进程内 fingerprint 抑制,不写持久化冷却;通知基础设施的 `notification_noise.py` 降噪仍独立生效。Web 规则列表使用后端返回的 `cooldown_active` 判断冷却状态,避免浏览器本地时区解析影响展示。
技术指标规则只使用日线 close 的边缘触发,partial bar 处理是服务器本地时区 + 16:00 的启发式,不做市场日历精确判定。WebUI 的“告警”页面可以管理持久化规则、执行一次性 dry-run 测试,并查看触发历史、通知尝试结果和只读冷却状态;详细边界见 [实时告警中心](alerts.md)。
技术指标规则只使用日线 close 的边缘触发,partial bar 处理是服务器本地时区 + 16:00 的启发式,不做市场日历精确判定。`watchlist` 每轮刷新 `STOCK_LIST` 后展开,`portfolio_holdings` 从持仓快照的非零持仓按 symbol 去重展开,`portfolio_account` 复用持仓风险服务做账户级聚合评估。WebUI 的“告警”页面可以管理持久化规则、执行一次性 dry-run 测试,并查看触发历史、通知尝试结果和只读冷却状态;批量规则的列表冷却状态是父规则摘要,子目标冷却以触发历史为准。详细边界见 [实时告警中心](alerts.md)。
## 持仓管理说明
+6 -2
View File
@@ -1160,7 +1160,7 @@ A: Check if Actions is enabled, and if cron expression is correct (note it's UTC
## Agent Event Monitor
When `AGENT_EVENT_MONITOR_ENABLED=true`, schedule mode runs the alert worker every `AGENT_EVENT_MONITOR_INTERVAL_MINUTES` minutes. The worker reads enabled rules created through the Alert API and continues to support legacy rules in `AGENT_EVENT_ALERT_RULES_JSON`; triggered alerts still go through the existing notification channels. Alert API / Web persisted rules support price, change-percent, volume, and daily technical indicator rules; legacy JSON still supports only the three basic rule types.
When `AGENT_EVENT_MONITOR_ENABLED=true`, schedule mode runs the alert worker every `AGENT_EVENT_MONITOR_INTERVAL_MINUTES` minutes. The worker reads enabled rules created through the Alert API and continues to support legacy rules in `AGENT_EVENT_ALERT_RULES_JSON`; triggered alerts still go through the existing notification channels. Alert API / Web persisted rules support price, change-percent, volume, daily technical indicators, and the `watchlist`, `portfolio_holdings`, and `portfolio_account` target scopes; legacy JSON still supports only the three basic rule types.
> Compatibility and rollback note: this section documents current Event Monitor rule behavior (including `price_change_percent`) and does not change external model/provider API semantics such as model names, providers, Base URL, LiteLLM, `OPENAI_*`, `DEEPSEEK_*`, or `GEMINI_*` configuration.
> Legacy JSON is not automatically migrated, deleted, or rewritten. To roll back the background alert worker, clear or disable `AGENT_EVENT_MONITOR_ENABLED`/related rule config.
@@ -1175,6 +1175,10 @@ When `AGENT_EVENT_MONITOR_ENABLED=true`, schedule mode runs the alert worker eve
| `macd_cross` | `bullish_cross` / `bearish_cross` | `fast_period`, `slow_period`, `signal_period` | DIF/DEA edge golden/death cross |
| `kdj_cross` | `bullish_cross` / `bearish_cross` | `period`, `k_period`, `d_period` | K/D edge golden/death cross |
| `cci_threshold` | `above` / `below` | `period`, `threshold` | CCI edge-crosses a threshold |
| `portfolio_stop_loss` | `mode=near|breach` | - | Account-level stop-loss proximity or breach |
| `portfolio_concentration` | - | - | Account-level symbol concentration |
| `portfolio_drawdown` | - | - | Account-level maximum drawdown alert |
| `portfolio_price_stale` | - | - | Stale or missing portfolio prices |
Example:
@@ -1186,7 +1190,7 @@ AGENT_EVENT_ALERT_RULES_JSON=[{"stock_code":"600519","alert_type":"price_cross",
The worker writes `triggered`, `skipped`, `degraded`, and `failed` rows to `alert_triggers` as evaluation history; normal non-triggered checks do not write history. For DB-persisted rules, `triggered` history is best-effort deduplicated by `rule_id + target + data_source + data_timestamp`: repeated hits for the same data point reuse the earliest trigger row, while records without `data_timestamp` are not deduplicated. Real triggers write per-channel attempts to `alert_notifications`, and Alert API persisted rules write business cooldown state to `alert_cooldowns`; if the persisted cooldown read fails, the worker temporarily falls back to the in-process fingerprint guard to avoid repeated notifications during the DB failure. Legacy `AGENT_EVENT_ALERT_RULES_JSON` rules continue to use the in-process fingerprint suppressor and do not write persisted cooldown state; the notification infrastructure `notification_noise.py` guard remains independent. The Web rule list uses the backend-provided `cooldown_active` flag instead of browser-local timezone parsing to decide whether a rule is cooling down.
Technical indicator rules use daily-close edge triggers only. Partial-bar handling is a server-local-time + 16:00 heuristic and does not implement market-calendar precision. The WebUI "Alerts" page can manage persisted rules, run one-shot dry-run tests, and view trigger history, notification attempts, and read-only cooldown state. See [Real-Time Alert Center](alerts.md) for detailed boundaries.
Technical indicator rules use daily-close edge triggers only. Partial-bar handling is a server-local-time + 16:00 heuristic and does not implement market-calendar precision. `watchlist` rules refresh and expand `STOCK_LIST` each worker run, `portfolio_holdings` expands non-zero snapshot positions with symbol de-duplication, and `portfolio_account` reuses the portfolio risk service for account-level aggregate evaluation. The WebUI "Alerts" page can manage persisted rules, run one-shot dry-run tests, and view trigger history, notification attempts, and read-only cooldown state; cooldown on batch rules is a parent-rule summary, while child-target cooldown details are visible through trigger history. See [Real-Time Alert Center](alerts.md) for detailed boundaries.
---
+289 -13
View File
@@ -8,7 +8,7 @@ import json
import logging
import re
from datetime import date, datetime
from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional
from src.agent.events import (
EventMonitor,
@@ -27,6 +27,26 @@ from src.services.alert_indicators import (
normalize_indicator_parameters,
threshold_for_indicator,
)
from src.services.portfolio_alerts import (
DRY_RUN_TARGET_TIMEOUT_SECONDS,
DRY_RUN_TOTAL_TIMEOUT_SECONDS,
PORTFOLIO_ALERT_TYPES,
SYMBOL_BATCH_TARGET_SCOPES,
PortfolioRiskAlert,
RuntimeAlertPayload,
StaticAlertEvaluation,
aggregate_dry_run_results,
ensure_active_portfolio_account,
evaluate_portfolio_risk_alert,
evaluate_static_alert,
expand_symbol_targets,
make_portfolio_risk_payload,
make_static_payload,
normalize_batch_target_scope_target,
normalize_portfolio_alert_parameters,
portfolio_effective_target,
result_to_target_result,
)
from src.storage import (
AlertCooldownRecord,
AlertNotificationRecord,
@@ -38,8 +58,9 @@ from src.utils.sanitize import sanitize_diagnostic_text
LEGACY_RUNTIME_ALERT_TYPES = frozenset({"price_cross", "price_change_percent", "volume_spike"})
SUPPORTED_ALERT_TYPES = LEGACY_RUNTIME_ALERT_TYPES | TECHNICAL_ALERT_TYPES
SUPPORTED_TARGET_SCOPES = frozenset({"single_symbol"})
SYMBOL_ALERT_TYPES = LEGACY_RUNTIME_ALERT_TYPES | TECHNICAL_ALERT_TYPES
SUPPORTED_ALERT_TYPES = SYMBOL_ALERT_TYPES | PORTFOLIO_ALERT_TYPES
SUPPORTED_TARGET_SCOPES = frozenset({"single_symbol", "watchlist", "portfolio_holdings", "portfolio_account"})
SUPPORTED_SEVERITIES = frozenset({"info", "warning", "critical"})
NULLABLE_RULE_UPDATE_FIELDS = frozenset({"cooldown_policy", "notification_policy"})
@@ -138,23 +159,33 @@ class AlertService:
if row is None:
raise AlertNotFoundError(f"Alert rule not found: {rule_id}")
rule = self._to_runtime_rule(row)
payloads = self.build_runtime_payloads(row)
monitor = EventMonitor()
try:
return asyncio.run(self._evaluate_rule(rule, monitor, daily_cache=None))
if len(payloads) == 1 and row.target_scope == "single_symbol":
result = asyncio.run(self._evaluate_rule(payloads[0].rule, monitor, daily_cache=None))
return self._dry_run_response_for_single(payloads[0], result, target_scope=row.target_scope)
results = asyncio.run(self._evaluate_runtime_payloads(payloads, monitor))
return aggregate_dry_run_results(rule_id, row.target_scope, results)
except Exception as exc:
sanitized_message = self._sanitize_text(str(exc) or "Alert evaluation failed")
return {
"rule_id": rule_id,
"target_scope": row.target_scope,
"status": "evaluation_error",
"record_status": "failed",
"triggered": False,
"observed_value": None,
"threshold": self._threshold_for_rule(rule),
"data_source": self._data_source_for_rule(rule),
"threshold": None,
"data_source": None,
"data_timestamp": None,
"reason": sanitized_message,
"message": sanitized_message,
"evaluated_count": 0,
"triggered_count": 0,
"degraded_count": 0,
"skipped_count": 0,
"target_results": [],
}
async def _evaluate_rule(
@@ -171,8 +202,95 @@ class AlertService:
return await self._evaluate_volume(rule)
if isinstance(rule, TechnicalIndicatorAlert):
return await self._evaluate_technical_indicator(rule, daily_cache=daily_cache)
if isinstance(rule, PortfolioRiskAlert):
return await asyncio.to_thread(evaluate_portfolio_risk_alert, rule)
if isinstance(rule, StaticAlertEvaluation):
return evaluate_static_alert(rule)
return self._evaluation_error(rule, f"unsupported runtime alert type: {rule.alert_type}")
async def _evaluate_runtime_payloads(
self,
payloads: List[RuntimeAlertPayload],
monitor: EventMonitor,
) -> List[Dict[str, Any]]:
semaphore = asyncio.Semaphore(8)
daily_cache: Dict[tuple[str, int], Any] = {}
async def _evaluate_one(payload: RuntimeAlertPayload) -> Dict[str, Any]:
async with semaphore:
try:
result = await asyncio.wait_for(
self._evaluate_rule(payload.rule, monitor, daily_cache=daily_cache),
timeout=DRY_RUN_TARGET_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
result = {
"rule_id": self._runtime_rule_id(payload.rule),
"status": "not_triggered",
"record_status": "skipped",
"triggered": False,
"observed_value": None,
"threshold": self._threshold_for_rule(payload.rule),
"data_source": self._data_source_for_rule(payload.rule),
"data_timestamp": None,
"reason": "dry-run evaluation timed out",
"message": "dry-run evaluation timed out",
}
except Exception as exc:
sanitized_message = self._sanitize_text(str(exc) or "Alert evaluation failed")
result = {
"rule_id": self._runtime_rule_id(payload.rule),
"status": "evaluation_error",
"record_status": "failed",
"triggered": False,
"observed_value": None,
"threshold": self._threshold_for_rule(payload.rule),
"data_source": self._data_source_for_rule(payload.rule),
"data_timestamp": None,
"reason": sanitized_message,
"message": sanitized_message,
}
return result_to_target_result(payload, result)
tasks = [asyncio.create_task(_evaluate_one(payload)) for payload in payloads]
done, pending = await asyncio.wait(tasks, timeout=DRY_RUN_TOTAL_TIMEOUT_SECONDS)
for task in pending:
task.cancel()
output: List[Dict[str, Any]] = []
for task in done:
output.append(task.result())
for task, payload in zip(tasks, payloads):
if task in pending:
output.append({
"target": payload.effective_target,
"display_target": payload.display_target,
"status": "not_triggered",
"record_status": "skipped",
"triggered": False,
"observed_value": None,
"threshold": None,
"message": "dry-run evaluation timed out",
})
return output
@staticmethod
def _dry_run_response_for_single(payload: RuntimeAlertPayload, result: Dict[str, Any], *, target_scope: str) -> Dict[str, Any]:
target_result = result_to_target_result(payload, result)
response = {
"rule_id": result.get("rule_id") or 0,
"target_scope": target_scope,
"status": result.get("status") or "evaluation_error",
"triggered": bool(result.get("triggered")),
"observed_value": result.get("observed_value"),
"message": result.get("message") or result.get("reason") or "",
"evaluated_count": 1,
"triggered_count": 1 if result.get("triggered") else 0,
"degraded_count": 1 if result.get("record_status") == "degraded" else 0,
"skipped_count": 1 if result.get("record_status") == "skipped" else 0,
"target_results": [target_result],
}
return response
async def _evaluate_price(self, rule: PriceAlert, monitor: EventMonitor) -> Dict[str, Any]:
threshold = float(rule.price)
try:
@@ -560,6 +678,8 @@ class AlertService:
return abs(float(rule.change_pct))
if isinstance(rule, TechnicalIndicatorAlert):
return threshold_for_indicator(rule.alert_type, rule.indicator_params)
if isinstance(rule, PortfolioRiskAlert):
return None
return None
@staticmethod
@@ -570,6 +690,8 @@ class AlertService:
return "daily_data"
if isinstance(rule, TechnicalIndicatorAlert):
return "daily_data"
if isinstance(rule, PortfolioRiskAlert):
return "portfolio_risk"
return None
@classmethod
@@ -734,13 +856,15 @@ class AlertService:
alert_type = str(payload.get("alert_type") or "").strip().lower()
if alert_type not in SUPPORTED_ALERT_TYPES:
raise UnsupportedAlertTypeError(f"unsupported alert_type for Alert API: {alert_type or '<empty>'}")
self._validate_scope_alert_type(target_scope, alert_type)
severity = str(payload.get("severity") or "warning").strip().lower()
if severity not in SUPPORTED_SEVERITIES:
raise AlertServiceError(f"unsupported severity: {severity}")
parameters = self._normalize_parameters(alert_type, payload.get("parameters") or {})
if alert_type in LEGACY_RUNTIME_ALERT_TYPES:
target = self._normalize_target(target_scope, target)
if target_scope == "single_symbol" and alert_type in LEGACY_RUNTIME_ALERT_TYPES:
serialized_rule = {"stock_code": target, "alert_type": alert_type, **parameters}
try:
validate_event_alert_rule(serialized_rule)
@@ -769,6 +893,28 @@ class AlertService:
if value is None and field_name not in NULLABLE_RULE_UPDATE_FIELDS:
raise AlertServiceError(f"{field_name} must not be null")
@staticmethod
def _validate_scope_alert_type(target_scope: str, alert_type: str) -> None:
if target_scope == "portfolio_account":
if alert_type not in PORTFOLIO_ALERT_TYPES:
raise AlertServiceError("portfolio_account only supports portfolio alert types")
return
if alert_type in PORTFOLIO_ALERT_TYPES:
raise AlertServiceError("portfolio alert types require target_scope=portfolio_account")
if target_scope in {"single_symbol", "watchlist", "portfolio_holdings"} and alert_type not in SYMBOL_ALERT_TYPES:
raise UnsupportedAlertTypeError(f"unsupported alert_type for {target_scope}: {alert_type}")
def _normalize_target(self, target_scope: str, target: str) -> str:
if target_scope == "single_symbol":
return target.strip()
try:
normalized = normalize_batch_target_scope_target(target_scope, target)
if target_scope in {"portfolio_holdings", "portfolio_account"}:
ensure_active_portfolio_account(normalized)
return normalized
except ValueError as exc:
raise AlertServiceError(str(exc)) from exc
def _normalize_parameters(self, alert_type: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
if not isinstance(parameters, dict):
raise AlertServiceError("parameters must be an object")
@@ -797,6 +943,12 @@ class AlertService:
except ValueError as exc:
raise AlertServiceError(str(exc)) from exc
if alert_type in PORTFOLIO_ALERT_TYPES:
try:
return normalize_portfolio_alert_parameters(alert_type, parameters)
except ValueError as exc:
raise AlertServiceError(str(exc)) from exc
raise UnsupportedAlertTypeError(f"unsupported alert_type for Alert API: {alert_type}")
@staticmethod
@@ -809,38 +961,149 @@ class AlertService:
raise AlertServiceError(f"{field_name} must be > 0")
return number
def build_runtime_payloads(
self,
row: AlertRuleRecord,
*,
config: Optional[Any] = None,
include_overflow_payload: bool = True,
) -> List[RuntimeAlertPayload]:
data = self._serialize_rule_base(row)
parent_key = self._semantic_key(
data["target_scope"],
data["target"],
data["alert_type"],
data["parameters"],
)
if data["alert_type"] in PORTFOLIO_ALERT_TYPES:
return [make_portfolio_risk_payload(parent_key=parent_key, data=data)]
if data["target_scope"] in SYMBOL_BATCH_TARGET_SCOPES:
if config is None:
from src.config import get_config
config = get_config()
try:
targets, overflow_count = expand_symbol_targets(
target_scope=data["target_scope"],
target=data["target"],
config=config,
)
except Exception as exc:
return [
make_static_payload(
parent_key=parent_key,
rule_id=int(data["id"] or 0),
alert_type=data["alert_type"],
effective_target=f"{data['target_scope']}:{data['target']}",
display_target=f"{data['target_scope']} {data['target']}",
message=self._sanitize_text(str(exc) or "target expansion failed"),
record_status="failed",
)
]
payloads: List[RuntimeAlertPayload] = []
for target in targets:
child_data = dict(data)
child_data["target"] = target.symbol
rule = self._to_runtime_rule(row, child_data)
effective_target = target.symbol
payloads.append(
RuntimeAlertPayload(
key=f"{parent_key}|{effective_target}",
rule=rule,
effective_target=effective_target,
display_target=target.display_target,
)
)
if overflow_count:
if include_overflow_payload:
payloads.append(
make_static_payload(
parent_key=parent_key,
rule_id=int(data["id"] or 0),
alert_type=data["alert_type"],
effective_target=f"{data['target_scope']}:{data['target']}:overflow",
display_target="展开目标超限",
message=f"Skipped {overflow_count} targets over soft cap",
record_status="degraded",
)
)
logger.warning(
"[AlertService] Alert rule %s expansion exceeded soft cap by %s targets",
data["id"],
overflow_count,
)
if not payloads:
scope_label = "watchlist" if data["target_scope"] == "watchlist" else "portfolio holdings"
payloads.append(
make_static_payload(
parent_key=parent_key,
rule_id=int(data["id"] or 0),
alert_type=data["alert_type"],
effective_target=f"{data['target_scope']}:{data['target']}",
display_target=scope_label,
message=f"No {scope_label} targets to evaluate",
record_status="skipped",
)
)
return payloads
rule = self._to_runtime_rule(row, data)
effective_target = str(data["target"])
return [
RuntimeAlertPayload(
key=parent_key,
rule=rule,
effective_target=effective_target,
display_target=effective_target,
)
]
def _to_runtime_rule(self, row: AlertRuleRecord, data: Optional[Dict[str, Any]] = None):
data = data or self._serialize_rule_base(row)
parameters = data["parameters"]
metadata = {
"persisted_rule_id": data["id"],
"target_scope": data.get("target_scope"),
"parent_target": row.target,
"effective_target": data.get("target"),
}
if data["alert_type"] == "price_cross":
return PriceAlert(
stock_code=data["target"],
direction=str(parameters["direction"]),
price=float(parameters["price"]),
metadata={"persisted_rule_id": data["id"]},
metadata=metadata,
)
if data["alert_type"] == "price_change_percent":
return PriceChangeAlert(
stock_code=data["target"],
direction=str(parameters["direction"]),
change_pct=float(parameters["change_pct"]),
metadata={"persisted_rule_id": data["id"]},
metadata=metadata,
)
if data["alert_type"] == "volume_spike":
return VolumeAlert(
stock_code=data["target"],
multiplier=float(parameters["multiplier"]),
metadata={"persisted_rule_id": data["id"]},
metadata=metadata,
)
if data["alert_type"] in TECHNICAL_ALERT_TYPES:
return TechnicalIndicatorAlert(
stock_code=data["target"],
alert_type=data["alert_type"],
indicator_params=parameters,
metadata={"persisted_rule_id": data["id"]},
metadata=metadata,
)
raise UnsupportedAlertTypeError(f"unsupported alert_type for Alert API: {data['alert_type']}")
@staticmethod
def _semantic_key(target_scope: str, target: str, alert_type: str, parameters: Dict[str, Any]) -> str:
canonical_params = json.dumps(parameters or {}, ensure_ascii=False, sort_keys=True)
return f"{target_scope}:{target}:{alert_type}:{canonical_params}"
def _serialize_rule(self, row: AlertRuleRecord) -> Dict[str, Any]:
data = self._serialize_rule_base(row)
cooldown_summary = self._cooldown_summary_for_rule(row)
@@ -870,9 +1133,14 @@ class AlertService:
def _cooldown_summary_for_rule(self, row: AlertRuleRecord) -> Dict[str, Any]:
try:
cooldown_target = (
portfolio_effective_target(str(row.target))
if str(row.target_scope) == "portfolio_account"
else str(row.target)
)
cooldown = self.repo.get_rule_cooldown_summary(
rule_id=int(row.id),
target=str(row.target),
target=cooldown_target,
severity=str(row.severity) if row.severity else None,
)
except Exception as exc:
@@ -946,6 +1214,14 @@ class AlertService:
return f"{target} KDJ {parameters['direction']}"
if alert_type == "cci_threshold":
return f"{target} CCI{parameters['period']} {parameters['direction']} {parameters['threshold']}"
if alert_type == "portfolio_stop_loss":
return f"{target} portfolio stop loss {parameters.get('mode', 'near')}"
if alert_type == "portfolio_concentration":
return f"{target} portfolio concentration"
if alert_type == "portfolio_drawdown":
return f"{target} portfolio drawdown"
if alert_type == "portfolio_price_stale":
return f"{target} portfolio stale price"
return f"{target} {alert_type}"
@staticmethod
+40 -25
View File
@@ -39,6 +39,8 @@ class RuntimeAlertRule:
source: str
severity: Optional[str] = None
cooldown_policy: Optional[Dict[str, Any]] = None
effective_target: Optional[str] = None
display_target: Optional[str] = None
@dataclass
@@ -179,23 +181,28 @@ class AlertWorker:
for row in self.service.repo.list_enabled_rules(limit=ALERT_WORKER_RULE_LIMIT):
try:
rule_data = self.service._serialize_rule_base(row)
key = self._semantic_key(
rule_data["target_scope"],
rule_data["target"],
rule_data["alert_type"],
rule_data["parameters"],
)
runtime_rules.append(
RuntimeAlertRule(
key=key,
rule=self.service._to_runtime_rule(row, rule_data),
source="db",
severity=rule_data.get("severity"),
cooldown_policy=rule_data.get("cooldown_policy"),
cooldown_policy = self.service._load_json(row.cooldown_policy, default=None)
for payload in self.service.build_runtime_payloads(row, config=config, include_overflow_payload=False):
if len(runtime_rules) >= ALERT_WORKER_RULE_LIMIT:
logger.warning(
"[AlertWorker] Runtime rule limit reached at %s; skipping remaining expanded rules",
ALERT_WORKER_RULE_LIMIT,
)
break
runtime_rules.append(
RuntimeAlertRule(
key=payload.key,
rule=payload.rule,
source="db",
severity=row.severity,
cooldown_policy=cooldown_policy,
effective_target=payload.effective_target,
display_target=payload.display_target,
)
)
)
seen_keys.add(key)
seen_keys.add(payload.key)
if len(runtime_rules) >= ALERT_WORKER_RULE_LIMIT:
break
except Exception as exc:
logger.warning("[AlertWorker] Skip invalid persisted alert rule %s: %s", getattr(row, "id", "?"), exc)
@@ -265,7 +272,7 @@ class AlertWorker:
fields = {
"rule_id": rule_id,
"target": runtime_rule.rule.stock_code,
"target": self._effective_target(runtime_rule),
"observed_value": self._optional_float(result.get("observed_value")),
"threshold": self._optional_float(result.get("threshold")),
"reason": result.get("reason") or result.get("message"),
@@ -293,7 +300,7 @@ class AlertWorker:
except Exception as exc:
logger.warning(
"[AlertWorker] Failed to record alert trigger for %s: %s",
getattr(runtime_rule.rule, "stock_code", "?"),
self._display_target(runtime_rule),
self.service._sanitize_text(str(exc) or "trigger write failed"),
)
return TriggerWriteResult()
@@ -319,7 +326,7 @@ class AlertWorker:
@staticmethod
def _diagnostics_for_status(status: str, result: Dict[str, Any]) -> Optional[str]:
if status == "triggered":
return None
return result.get("diagnostics")
return result.get("message") or result.get("reason")
def _should_notify(self, rule_key: str, *, ttl_seconds: Optional[int] = None) -> bool:
@@ -361,7 +368,7 @@ class AlertWorker:
from src.notification import NotificationBuilder, NotificationService
notification_service = self.notifier or NotificationService()
title = f"Event Alert | {runtime_rule.rule.stock_code}"
title = f"Event Alert | {self._display_target(runtime_rule)}"
content = result.get("reason") or result.get("message") or runtime_rule.rule.description or "Alert triggered"
alert_text = NotificationBuilder.build_simple_alert(title=title, content=content, alert_type="warning")
@@ -376,7 +383,7 @@ class AlertWorker:
sanitized = self.service._sanitize_text(str(exc) or "notification failed")
logger.warning(
"[AlertWorker] Failed to send alert notification for %s: %s",
getattr(runtime_rule.rule, "stock_code", "?"),
self._display_target(runtime_rule),
sanitized,
)
return NotificationDispatchResult(
@@ -487,14 +494,14 @@ class AlertWorker:
try:
cooldown = self.service.repo.get_active_cooldown(
rule_id=rule_id,
target=runtime_rule.rule.stock_code,
target=self._effective_target(runtime_rule),
severity=runtime_rule.severity,
now=now_dt,
)
except Exception as exc:
logger.warning(
"[AlertWorker] Failed to read alert cooldown for %s: %s",
getattr(runtime_rule.rule, "stock_code", "?"),
self._display_target(runtime_rule),
self.service._sanitize_text(str(exc) or "cooldown read failed"),
)
fallback_key = self._db_cooldown_fallback_key(runtime_rule.key)
@@ -570,7 +577,7 @@ class AlertWorker:
self.service.repo.upsert_cooldown(
rule_id=rule_id,
rule_key=runtime_rule.key,
target=runtime_rule.rule.stock_code,
target=self._effective_target(runtime_rule),
severity=runtime_rule.severity,
last_triggered_at=now_dt,
cooldown_until=now_dt + timedelta(seconds=cooldown_seconds),
@@ -579,10 +586,18 @@ class AlertWorker:
except Exception as exc:
logger.warning(
"[AlertWorker] Failed to update alert cooldown for %s: %s",
getattr(runtime_rule.rule, "stock_code", "?"),
self._display_target(runtime_rule),
self.service._sanitize_text(str(exc) or "cooldown write failed"),
)
@staticmethod
def _effective_target(runtime_rule: RuntimeAlertRule) -> str:
return str(runtime_rule.effective_target or getattr(runtime_rule.rule, "stock_code", "") or "?")
@staticmethod
def _display_target(runtime_rule: RuntimeAlertRule) -> str:
return str(runtime_rule.display_target or runtime_rule.effective_target or getattr(runtime_rule.rule, "stock_code", "") or "?")
@staticmethod
def _cooldown_seconds(runtime_rule: RuntimeAlertRule) -> int:
policy = runtime_rule.cooldown_policy if isinstance(runtime_rule.cooldown_policy, dict) else None
+616
View File
@@ -0,0 +1,616 @@
# -*- coding: utf-8 -*-
"""Portfolio and watchlist alert helpers for Alert Center P6."""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any, Dict, Iterable, List, Optional
from src.services.portfolio_risk_service import PortfolioRiskService
from src.services.portfolio_service import PortfolioService
logger = logging.getLogger(__name__)
SYMBOL_BATCH_TARGET_SCOPES = frozenset({"watchlist", "portfolio_holdings"})
PORTFOLIO_TARGET_SCOPES = frozenset({"portfolio_holdings", "portfolio_account"})
PORTFOLIO_ALERT_TYPES = frozenset({
"portfolio_stop_loss",
"portfolio_concentration",
"portfolio_drawdown",
"portfolio_price_stale",
})
EXPANDED_TARGET_SOFT_CAP = 100
TARGET_RESULTS_LIMIT = 20
DRY_RUN_TARGET_TIMEOUT_SECONDS = 10
DRY_RUN_TOTAL_TIMEOUT_SECONDS = 30
@dataclass(frozen=True)
class ExpandedSymbolTarget:
"""A concrete symbol produced from a parent batch rule."""
symbol: str
display_target: str
@dataclass(frozen=True)
class RuntimeAlertPayload:
"""Runtime rule plus the identity used for cooldown/history."""
key: str
rule: Any
effective_target: str
display_target: str
@dataclass
class PortfolioRiskAlert:
"""Runtime alert for account-level portfolio risk rules."""
target_scope: str
target: str
alert_type: str
parameters: Dict[str, Any]
metadata: Dict[str, Any] = field(default_factory=dict)
description: str = ""
stock_code: str = ""
def __post_init__(self) -> None:
effective_target = self.metadata.get("effective_target") or portfolio_effective_target(self.target)
self.stock_code = str(effective_target)
@dataclass
class StaticAlertEvaluation:
"""Runtime placeholder for skipped/degraded expansion results."""
stock_code: str
alert_type: str
message: str
record_status: str = "skipped"
metadata: Dict[str, Any] = field(default_factory=dict)
description: str = ""
def normalize_portfolio_alert_parameters(alert_type: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize P6 portfolio alert parameters."""
if alert_type not in PORTFOLIO_ALERT_TYPES:
raise ValueError(f"unsupported portfolio alert_type: {alert_type}")
if not isinstance(parameters, dict):
raise ValueError("parameters must be an object")
if alert_type == "portfolio_stop_loss":
mode = str(parameters.get("mode") or "near").strip().lower()
if mode not in {"near", "breach"}:
raise ValueError("portfolio_stop_loss mode must be near or breach")
return {"mode": mode}
return {}
def portfolio_effective_target(target: str) -> str:
target_text = str(target or "all").strip() or "all"
return "account:all" if target_text == "all" else f"account:{target_text}"
def normalize_batch_target_scope_target(target_scope: str, target: str) -> str:
target_text = str(target or "").strip()
if target_scope == "watchlist":
if target_text not in {"", "default"}:
raise ValueError("watchlist target must be default")
return "default"
if target_scope in PORTFOLIO_TARGET_SCOPES:
if target_text == "all":
return "all"
return str(_positive_int_target(target_text))
return target_text
def ensure_active_portfolio_account(target: str, *, portfolio_service: Optional[PortfolioService] = None) -> None:
"""Validate that an explicit portfolio account target exists and is active."""
if str(target or "").strip() == "all":
return
account_id = _positive_int_target(target)
service = portfolio_service or PortfolioService()
accounts = service.list_accounts(include_inactive=False)
active_ids = {int(item.get("id")) for item in accounts if item.get("id") is not None}
if account_id not in active_ids:
raise ValueError(f"portfolio account is not active or does not exist: {account_id}")
def expand_symbol_targets(
*,
target_scope: str,
target: str,
config: Any,
portfolio_service: Optional[PortfolioService] = None,
) -> tuple[List[ExpandedSymbolTarget], int]:
"""Expand watchlist or portfolio holdings into concrete, de-duplicated symbols.
Returns ``(targets, overflow_count)``. The returned targets are already capped
by ``EXPANDED_TARGET_SOFT_CAP``.
"""
if target_scope == "watchlist":
symbols = _watchlist_symbols(config)
display_prefix = "自选股"
elif target_scope == "portfolio_holdings":
symbols = _portfolio_holding_symbols(target=target, portfolio_service=portfolio_service)
display_prefix = "持仓"
else:
return [], 0
unique = _dedupe_symbols(symbols)
overflow_count = max(0, len(unique) - EXPANDED_TARGET_SOFT_CAP)
capped = unique[:EXPANDED_TARGET_SOFT_CAP]
return [
ExpandedSymbolTarget(symbol=symbol, display_target=f"{display_prefix} - {symbol}")
for symbol in capped
], overflow_count
def make_static_payload(
*,
parent_key: str,
rule_id: int,
alert_type: str,
effective_target: str,
display_target: str,
message: str,
record_status: str = "skipped",
) -> RuntimeAlertPayload:
rule = StaticAlertEvaluation(
stock_code=effective_target,
alert_type=alert_type,
message=message,
record_status=record_status,
metadata={
"persisted_rule_id": rule_id,
"effective_target": effective_target,
"display_target": display_target,
},
description=message,
)
return RuntimeAlertPayload(
key=f"{parent_key}|{effective_target}",
rule=rule,
effective_target=effective_target,
display_target=display_target,
)
def make_portfolio_risk_payload(
*,
parent_key: str,
data: Dict[str, Any],
) -> RuntimeAlertPayload:
effective_target = portfolio_effective_target(data["target"])
display_target = "全部账户" if data["target"] == "all" else f"账户 {data['target']}"
rule = PortfolioRiskAlert(
target_scope=data["target_scope"],
target=data["target"],
alert_type=data["alert_type"],
parameters=dict(data.get("parameters") or {}),
metadata={
"persisted_rule_id": data["id"],
"effective_target": effective_target,
"display_target": display_target,
},
description=data.get("name") or data["alert_type"],
)
return RuntimeAlertPayload(
key=f"{parent_key}|{effective_target}",
rule=rule,
effective_target=effective_target,
display_target=display_target,
)
def evaluate_static_alert(rule: StaticAlertEvaluation) -> Dict[str, Any]:
return {
"rule_id": int(rule.metadata.get("persisted_rule_id", 0) or 0),
"status": "not_triggered",
"record_status": rule.record_status,
"triggered": False,
"observed_value": None,
"threshold": None,
"data_source": None,
"data_timestamp": None,
"reason": rule.message,
"message": rule.message,
}
def evaluate_portfolio_risk_alert(
rule: PortfolioRiskAlert,
*,
portfolio_service: Optional[PortfolioService] = None,
risk_service: Optional[PortfolioRiskService] = None,
) -> Dict[str, Any]:
"""Evaluate an account-level portfolio alert."""
account_id = None if rule.target == "all" else _positive_int_target(rule.target)
service = portfolio_service or PortfolioService()
risk = risk_service or PortfolioRiskService(portfolio_service=service)
if rule.alert_type == "portfolio_price_stale":
snapshot = service.get_portfolio_snapshot(account_id=account_id, cost_method="fifo")
return _evaluate_price_stale(rule, snapshot)
report = risk.get_risk_report(account_id=account_id, cost_method="fifo")
if rule.alert_type == "portfolio_stop_loss":
return _evaluate_stop_loss(rule, report)
if rule.alert_type == "portfolio_concentration":
return _evaluate_concentration(rule, report)
if rule.alert_type == "portfolio_drawdown":
return _evaluate_drawdown(rule, report)
return _portfolio_result(
rule,
triggered=False,
observed_value=None,
threshold=None,
message=f"unsupported portfolio alert_type: {rule.alert_type}",
record_status="failed",
diagnostics={"error": "unsupported_portfolio_alert_type"},
)
def result_to_target_result(payload: RuntimeAlertPayload, result: Dict[str, Any]) -> Dict[str, Any]:
record_status = result.get("record_status")
return {
"target": payload.effective_target,
"display_target": payload.display_target,
"status": result.get("status") or "evaluation_error",
"record_status": record_status,
"triggered": bool(result.get("triggered")),
"observed_value": result.get("observed_value"),
"threshold": result.get("threshold"),
"message": result.get("message") or result.get("reason") or "",
}
def aggregate_dry_run_results(rule_id: int, target_scope: str, results: List[Dict[str, Any]]) -> Dict[str, Any]:
target_results = sorted(
results,
key=lambda item: (
0 if item.get("triggered") else 1,
0 if item.get("record_status") in {"degraded", "failed"} else 1,
str(item.get("target") or ""),
),
)
visible_results = target_results[:TARGET_RESULTS_LIMIT]
triggered_count = sum(1 for item in target_results if item.get("triggered"))
degraded_count = sum(1 for item in target_results if item.get("record_status") == "degraded")
skipped_count = sum(1 for item in target_results if item.get("record_status") == "skipped")
failed_count = sum(1 for item in target_results if item.get("record_status") == "failed")
successful_count = sum(
1
for item in target_results
if item.get("record_status") not in {"failed"} and item.get("status") != "evaluation_error"
)
if triggered_count:
status = "triggered"
triggered = True
elif successful_count or skipped_count or degraded_count:
status = "not_triggered"
triggered = False
else:
status = "evaluation_error"
triggered = False
if not target_results:
status = "evaluation_error"
triggered = False
message = "No targets were evaluated"
else:
message = (
f"Evaluated {len(target_results)} targets: "
f"{triggered_count} triggered, {degraded_count} degraded, "
f"{skipped_count} skipped, {failed_count} failed"
)
first_observed = next((item.get("observed_value") for item in target_results if item.get("observed_value") is not None), None)
return {
"rule_id": rule_id,
"target_scope": target_scope,
"status": status,
"triggered": triggered,
"observed_value": first_observed,
"message": message,
"evaluated_count": len(target_results),
"triggered_count": triggered_count,
"degraded_count": degraded_count,
"skipped_count": skipped_count,
"target_results": visible_results,
}
def _watchlist_symbols(config: Any) -> List[str]:
refresh = getattr(config, "refresh_stock_list", None)
if callable(refresh):
try:
refresh()
except Exception as exc:
logger.warning("[portfolio_alerts] Failed to refresh watchlist symbols: %s", exc)
return list(getattr(config, "stock_list", []) or [])
def _portfolio_holding_symbols(
*,
target: str,
portfolio_service: Optional[PortfolioService],
) -> List[str]:
service = portfolio_service or PortfolioService()
account_id = None if target == "all" else _positive_int_target(target)
snapshot = service.get_portfolio_snapshot(account_id=account_id, cost_method="fifo")
symbols: List[str] = []
for account in snapshot.get("accounts", []) or []:
for position in account.get("positions", []) or []:
try:
quantity = float(position.get("quantity") or 0.0)
except (TypeError, ValueError):
quantity = 0.0
if quantity <= 0:
continue
symbol = _normalize_symbol(position.get("symbol"))
if symbol:
symbols.append(symbol)
return symbols
def _dedupe_symbols(symbols: Iterable[Any]) -> List[str]:
output: List[str] = []
seen = set()
for raw in symbols:
symbol = _normalize_symbol(raw)
if not symbol or symbol in seen:
continue
output.append(symbol)
seen.add(symbol)
return output
def _normalize_symbol(value: Any) -> str:
return PortfolioService._normalize_symbol(str(value or ""))
def _positive_int_target(value: Any) -> int:
try:
account_id = int(str(value).strip())
except (TypeError, ValueError) as exc:
raise ValueError("portfolio target must be all or a positive account id") from exc
if account_id <= 0:
raise ValueError("portfolio target must be all or a positive account id")
return account_id
def _evaluate_stop_loss(rule: PortfolioRiskAlert, report: Dict[str, Any]) -> Dict[str, Any]:
mode = str(rule.parameters.get("mode") or "near")
stop_loss = report.get("stop_loss") or {}
items = list(stop_loss.get("items") or [])
if mode == "breach":
affected = [item for item in items if bool(item.get("is_triggered"))]
triggered = bool(affected)
else:
affected = items
triggered = bool(stop_loss.get("near_alert")) and bool(affected)
threshold_key = "stop_loss_alert_pct" if mode == "breach" else "stop_loss_near_ratio"
threshold = _threshold(report, threshold_key)
if mode == "near":
stop_loss_pct = _threshold(report, "stop_loss_alert_pct") or 0.0
near_ratio = _threshold(report, "stop_loss_near_ratio") or 0.0
threshold = stop_loss_pct * near_ratio
observed = max((float(item.get("loss_pct") or 0.0) for item in affected), default=0.0)
diagnostics = _base_diagnostics(report, top_items=affected[:5])
diagnostics.update({
"mode": mode,
"near_count": stop_loss.get("near_count", 0),
"triggered_count": stop_loss.get("triggered_count", 0),
})
message = (
f"{_display_account(report)} stop-loss {mode}: {len(affected)} affected symbols"
if triggered
else f"{_display_account(report)} stop-loss {mode}: no affected symbols"
)
return _portfolio_result(
rule,
triggered=triggered,
observed_value=observed,
threshold=threshold,
message=message,
diagnostics=diagnostics,
)
def _evaluate_concentration(rule: PortfolioRiskAlert, report: Dict[str, Any]) -> Dict[str, Any]:
concentration = report.get("concentration") or {}
observed = float(concentration.get("top_weight_pct") or 0.0)
threshold = _threshold(report, "concentration_alert_pct")
triggered = bool(concentration.get("alert"))
diagnostics = _base_diagnostics(report, top_items=concentration.get("top_positions") or [])
diagnostics.update({
"total_market_value": concentration.get("total_market_value"),
"top_weight_pct": observed,
})
message = f"{_display_account(report)} concentration top weight {observed:.2f}%"
return _portfolio_result(
rule,
triggered=triggered,
observed_value=observed,
threshold=threshold,
message=message,
diagnostics=diagnostics,
)
def _evaluate_drawdown(rule: PortfolioRiskAlert, report: Dict[str, Any]) -> Dict[str, Any]:
drawdown = report.get("drawdown") or {}
observed = float(drawdown.get("max_drawdown_pct") or 0.0)
threshold = _threshold(report, "drawdown_alert_pct")
triggered = bool(drawdown.get("alert"))
diagnostics = _base_diagnostics(report)
diagnostics.update({
"series_points": drawdown.get("series_points"),
"current_drawdown_pct": drawdown.get("current_drawdown_pct"),
"max_drawdown_pct": observed,
"fx_stale": bool(drawdown.get("fx_stale")),
})
message = f"{_display_account(report)} max drawdown {observed:.2f}%"
return _portfolio_result(
rule,
triggered=triggered,
observed_value=observed,
threshold=threshold,
message=message,
diagnostics=diagnostics,
)
def _evaluate_price_stale(rule: PortfolioRiskAlert, snapshot: Dict[str, Any]) -> Dict[str, Any]:
affected: List[Dict[str, Any]] = []
for account in snapshot.get("accounts", []) or []:
for position in account.get("positions", []) or []:
if bool(position.get("price_stale")) or not bool(position.get("price_available", True)):
affected.append({
"account_id": account.get("account_id"),
"symbol": position.get("symbol"),
"price_stale": bool(position.get("price_stale")),
"price_available": bool(position.get("price_available")),
"price_source": position.get("price_source"),
"price_date": position.get("price_date"),
})
diagnostics = _base_diagnostics_from_snapshot(snapshot, top_items=affected[:5])
observed = float(len(affected))
message = (
f"{_display_snapshot_account(snapshot)} stale or missing prices: {len(affected)} symbols"
if affected
else f"{_display_snapshot_account(snapshot)} prices are current"
)
return _portfolio_result(
rule,
triggered=bool(affected),
observed_value=observed,
threshold=0.0,
message=message,
diagnostics=diagnostics,
data_timestamp=_parse_date(snapshot.get("as_of")),
data_source="portfolio_snapshot",
)
def _portfolio_result(
rule: PortfolioRiskAlert,
*,
triggered: bool,
observed_value: Optional[float],
threshold: Optional[float],
message: str,
diagnostics: Dict[str, Any],
record_status: Optional[str] = None,
data_timestamp: Optional[datetime] = None,
data_source: str = "portfolio_risk",
) -> Dict[str, Any]:
if data_timestamp is None:
data_timestamp = _parse_date(diagnostics.get("as_of"))
status = "triggered" if triggered else "not_triggered"
return {
"rule_id": int(rule.metadata.get("persisted_rule_id", 0) or 0),
"status": status,
"record_status": "triggered" if triggered else record_status,
"triggered": triggered,
"observed_value": observed_value,
"threshold": threshold,
"data_source": data_source,
"data_timestamp": data_timestamp,
"reason": message,
"message": message,
"diagnostics": json.dumps(diagnostics, ensure_ascii=False, sort_keys=True),
}
def _base_diagnostics(report: Dict[str, Any], *, top_items: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
return {
"account_id": report.get("account_id") if report.get("account_id") is not None else "all",
"currency": report.get("currency"),
"as_of": report.get("as_of"),
"price_stale": False,
"fx_stale": bool((report.get("drawdown") or {}).get("fx_stale")),
"data_available": True,
"top_affected_symbols": _top_symbols(top_items or []),
}
def _base_diagnostics_from_snapshot(
snapshot: Dict[str, Any],
*,
top_items: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
accounts = snapshot.get("accounts", []) or []
explicit_account = accounts[0].get("account_id") if len(accounts) == 1 else "all"
affected = top_items or []
return {
"account_id": explicit_account,
"currency": snapshot.get("currency"),
"as_of": snapshot.get("as_of"),
"price_stale": any(bool(item.get("price_stale")) for item in affected),
"fx_stale": bool(snapshot.get("fx_stale")),
"data_available": all(bool(item.get("price_available")) for item in affected) if affected else True,
"top_affected_symbols": _top_symbols(affected),
}
def _top_symbols(items: List[Dict[str, Any]]) -> List[str]:
output: List[str] = []
for item in items[:5]:
symbol = str(item.get("symbol") or "").strip()
if symbol:
output.append(symbol)
return output
def _threshold(report: Dict[str, Any], name: str) -> Optional[float]:
thresholds = report.get("thresholds") or {}
value = thresholds.get(name)
if value is None:
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def _display_account(report: Dict[str, Any]) -> str:
account_id = report.get("account_id")
return "account all" if account_id is None else f"account {account_id}"
def _display_snapshot_account(snapshot: Dict[str, Any]) -> str:
accounts = snapshot.get("accounts", []) or []
if len(accounts) == 1:
return f"account {accounts[0].get('account_id')}"
return "account all"
def _parse_date(value: Any) -> Optional[datetime]:
if value is None:
return None
if isinstance(value, datetime):
return value
if isinstance(value, date):
return datetime.combine(value, datetime.min.time())
try:
return datetime.fromisoformat(str(value))
except ValueError:
return None
+142
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import os
import sys
import tempfile
@@ -24,6 +25,7 @@ import src.auth as auth
from api.app import create_app
from src.config import Config
from src.repositories.alert_repo import AlertRepository
from src.services.portfolio_service import PortfolioService
from src.storage import AlertCooldownRecord, AlertNotificationRecord, AlertTriggerRecord, Base, DatabaseManager
@@ -309,6 +311,146 @@ class AlertApiTestCase(unittest.TestCase):
self.assertEqual(resp.status_code, 400, resp.text)
self.assertEqual(resp.json()["error"], "validation_error")
def test_p6_scope_type_matrix_and_target_validation(self) -> None:
account = PortfolioService().create_account(
name="Main",
broker="Demo",
market="us",
base_currency="USD",
)
valid_cases = [
{
"target_scope": "watchlist",
"target": "default",
"alert_type": "price_cross",
"parameters": {"direction": "above", "price": 10},
},
{
"target_scope": "portfolio_holdings",
"target": str(account["id"]),
"alert_type": "rsi_threshold",
"parameters": {"direction": "below", "period": 12, "threshold": 30},
},
{
"target_scope": "portfolio_account",
"target": "all",
"alert_type": "portfolio_stop_loss",
"parameters": {"mode": "breach"},
},
]
for body in valid_cases:
resp = self.client.post("/api/v1/alerts/rules", json=body)
self.assertEqual(resp.status_code, 200, resp.text)
self.assertEqual(resp.json()["target_scope"], body["target_scope"])
invalid_cases = [
{
"target_scope": "watchlist",
"target": "600519",
"alert_type": "price_cross",
"parameters": {"direction": "above", "price": 10},
},
{
"target_scope": "portfolio_account",
"target": "all",
"alert_type": "price_cross",
"parameters": {"direction": "above", "price": 10},
},
{
"target_scope": "portfolio_holdings",
"target": "all",
"alert_type": "portfolio_drawdown",
"parameters": {},
},
{
"target_scope": "portfolio_account",
"target": "99999",
"alert_type": "portfolio_drawdown",
"parameters": {},
},
]
for body in invalid_cases:
resp = self.client.post("/api/v1/alerts/rules", json=body)
self.assertEqual(resp.status_code, 400, resp.text)
self.assertEqual(resp.json()["error"], "validation_error")
def test_p6_watchlist_dry_run_aggregates_targets_without_stock_code_validation(self) -> None:
rule = self._create_rule({
"name": "Watchlist breakout",
"target_scope": "watchlist",
"target": "default",
"alert_type": "price_cross",
"parameters": {"direction": "above", "price": 10},
})
async def _quote(_monitor, stock_code):
return SimpleNamespace(price=11.0 if stock_code == "600519" else 9.0)
with patch("src.agent.events.EventMonitor._get_realtime_quote", new=_quote):
resp = self.client.post(f"/api/v1/alerts/rules/{rule['id']}/test")
self.assertEqual(resp.status_code, 200, resp.text)
payload = resp.json()
self.assertEqual(payload["target_scope"], "watchlist")
self.assertTrue(payload["triggered"])
self.assertGreaterEqual(payload["evaluated_count"], 1)
self.assertEqual(payload["triggered_count"], 1)
self.assertEqual(payload["target_results"][0]["target"], "600519")
def test_p6_watchlist_dry_run_timeout_counts_target_as_skipped(self) -> None:
rule = self._create_rule({
"name": "Watchlist slow",
"target_scope": "watchlist",
"target": "default",
"alert_type": "price_cross",
"parameters": {"direction": "above", "price": 10},
})
async def _slow_quote(_monitor, _stock_code):
await asyncio.sleep(0.05)
return SimpleNamespace(price=11.0)
with patch("src.services.alert_service.DRY_RUN_TARGET_TIMEOUT_SECONDS", 0.001), patch(
"src.agent.events.EventMonitor._get_realtime_quote",
new=_slow_quote,
):
resp = self.client.post(f"/api/v1/alerts/rules/{rule['id']}/test")
self.assertEqual(resp.status_code, 200, resp.text)
payload = resp.json()
self.assertEqual(payload["status"], "not_triggered")
self.assertFalse(payload["triggered"])
self.assertEqual(payload["evaluated_count"], 1)
self.assertEqual(payload["skipped_count"], 1)
self.assertEqual(payload["target_results"][0]["record_status"], "skipped")
self.assertIn("timed out", payload["target_results"][0]["message"])
def test_p6_portfolio_account_cooldown_summary_uses_effective_target(self) -> None:
created = self._create_rule({
"name": "Portfolio drawdown",
"target_scope": "portfolio_account",
"target": "all",
"alert_type": "portfolio_drawdown",
"parameters": {},
})
repo = AlertRepository(self.db)
now_dt = datetime.now()
cooldown_until = now_dt + timedelta(minutes=5)
repo.upsert_cooldown(
rule_id=created["id"],
rule_key="portfolio_account:all:portfolio_drawdown:{}|account:all",
target="account:all",
severity="warning",
last_triggered_at=now_dt,
cooldown_until=cooldown_until,
reason="active cooldown",
)
detail_resp = self.client.get(f"/api/v1/alerts/rules/{created['id']}")
self.assertEqual(detail_resp.status_code, 200, detail_resp.text)
self.assertTrue(detail_resp.json()["cooldown_active"])
def test_rejects_unsupported_and_invalid_rules(self) -> None:
unsupported = self.client.post(
"/api/v1/alerts/rules",
+137
View File
@@ -1365,6 +1365,143 @@ class AlertWorkerTestCase(unittest.TestCase):
self.assertEqual(notifier.send_with_results.call_count, 3)
self.assertEqual(len(self._triggers(status="triggered")), 4)
def test_p6_watchlist_expands_to_child_keys_for_db_cooldown_fallback(self) -> None:
self._create_rule(
name="Watchlist",
target_scope="watchlist",
target="default",
alert_type="price_cross",
parameters={"direction": "above", "price": 10},
cooldown_policy={"cooldown_seconds": 60},
)
notifier = self._notifier()
config = self._config()
config.stock_list = ["600519", "000001"]
now = {"value": 1000.0}
async def _quote(_monitor, _stock_code):
return SimpleNamespace(price=11.0)
worker = AlertWorker(
config_provider=lambda: config,
service=self.service,
notifier=notifier,
now_provider=lambda: now["value"],
fingerprint_ttl_seconds=86400,
)
with patch.object(
self.service.repo,
"get_active_cooldown",
side_effect=RuntimeError("database locked"),
), patch("src.agent.events.EventMonitor._get_realtime_quote", new=_quote):
first = worker.run_once()
now["value"] += 10
second = worker.run_once()
self.assertEqual(first["loaded"], 2)
self.assertEqual(first["notified"], 2)
self.assertEqual(second["cooldown_suppressed"], 2)
self.assertEqual(notifier.send_with_results.call_count, 2)
targets = {item["target"] for item in self._triggers(status="triggered")}
self.assertEqual(targets, {"600519", "000001"})
def test_p6_empty_watchlist_writes_skipped_trigger(self) -> None:
self._create_rule(
name="Watchlist",
target_scope="watchlist",
target="default",
alert_type="price_cross",
parameters={"direction": "above", "price": 10},
)
config = self._config()
config.stock_list = []
worker = AlertWorker(config_provider=lambda: config, service=self.service, notifier=self._notifier())
stats = worker.run_once()
self.assertEqual(stats["skipped"], 1)
triggers = self._triggers(status="skipped")
self.assertEqual(len(triggers), 1)
self.assertEqual(triggers[0]["target"], "watchlist:default")
self.assertIn("No watchlist targets", triggers[0]["diagnostics"])
def test_p6_overflow_payload_is_dry_run_only_and_worker_does_not_write_degraded_history(self) -> None:
rule = self._create_rule(
name="Large watchlist",
target_scope="watchlist",
target="default",
alert_type="price_cross",
parameters={"direction": "above", "price": 10},
)
row = self.service.repo.get_rule(rule["id"])
config = self._config()
config.stock_list = [f"{index:06d}" for index in range(1, 102)]
dry_run_payloads = self.service.build_runtime_payloads(row, config=config)
worker_payloads = self.service.build_runtime_payloads(row, config=config, include_overflow_payload=False)
self.assertEqual(len(dry_run_payloads), 101)
self.assertTrue(dry_run_payloads[-1].effective_target.endswith(":overflow"))
self.assertEqual(len(worker_payloads), 100)
self.assertFalse(any(payload.effective_target.endswith(":overflow") for payload in worker_payloads))
async def _not_triggered(rule_obj, *_args, **_kwargs):
return {
"rule_id": self.service._runtime_rule_id(rule_obj),
"status": "not_triggered",
"record_status": None,
"triggered": False,
"observed_value": 9.0,
"threshold": 10.0,
"data_source": "realtime_quote",
"data_timestamp": None,
"reason": "below threshold",
"message": "below threshold",
}
worker = AlertWorker(config_provider=lambda: config, service=self.service, notifier=self._notifier())
with patch.object(self.service, "_evaluate_rule", new=_not_triggered):
stats = worker.run_once()
self.assertEqual(stats["loaded"], 100)
self.assertEqual(stats["degraded"], 0)
self.assertEqual(self._triggers(status="degraded"), [])
def test_p6_portfolio_account_risk_uses_account_effective_target_and_diagnostics(self) -> None:
rule = self._create_rule(
name="Portfolio risk",
target_scope="portfolio_account",
target="all",
alert_type="portfolio_concentration",
parameters={},
)
notifier = self._notifier()
async def _evaluate_portfolio(rule_obj, *_args, **_kwargs):
return {
"rule_id": self.service._runtime_rule_id(rule_obj),
"status": "triggered",
"record_status": "triggered",
"triggered": True,
"observed_value": 42.0,
"threshold": 35.0,
"data_source": "portfolio_risk",
"data_timestamp": None,
"reason": "account all concentration top weight 42.00%",
"message": "account all concentration top weight 42.00%",
"diagnostics": '{"account_id":"all","currency":"CNY","as_of":"2026-05-20"}',
}
worker = AlertWorker(config_provider=lambda: self._config(), service=self.service, notifier=notifier)
with patch.object(self.service, "_evaluate_rule", new=_evaluate_portfolio):
stats = worker.run_once()
self.assertEqual(stats["triggered"], 1)
triggers = self._triggers(rule_id=rule["id"], status="triggered")
self.assertEqual(len(triggers), 1)
self.assertEqual(triggers[0]["target"], "account:all")
self.assertIn("account_id", triggers[0]["diagnostics"])
if __name__ == "__main__":
unittest.main()
+37
View File
@@ -218,3 +218,40 @@ def test_alerts_doc_defines_p5_indicator_scope() -> None:
"skip unsupported `alert_type`",
):
assert token in doc
def test_alerts_doc_defines_p6_portfolio_and_watchlist_scope() -> None:
doc = _read_doc()
for token in (
"## P6 持仓与自选股联动",
"P6 scope/type 矩阵",
"`watchlist`",
"`portfolio_holdings`",
"`portfolio_account`",
"`portfolio_stop_loss`",
"`portfolio_concentration`",
"`portfolio_drawdown`",
"`portfolio_price_stale`",
"Target Identity Contract",
"`effective_target`",
"`RuntimeAlertRule.key`",
"`{parent_key}|{effective_target}`",
"dry-run",
"`degraded_count`",
"soft cap",
"cooldown_active",
"父规则摘要",
"legacy `AGENT_EVENT_ALERT_RULES_JSON` 不支持 watchlist、portfolio",
"sector 级集中度",
"P6 PR",
):
assert token in doc
def test_changelog_unreleased_keeps_flat_entries_for_alert_p6() -> None:
changelog = (PROJECT_ROOT / "docs" / "CHANGELOG.md").read_text(encoding="utf-8")
unreleased = changelog.split("## [Unreleased]", 1)[1].split("\n## [", 1)[0]
assert "- [新功能] 告警中心 P6 支持自选股、持仓标的和持仓账户联动规则" in unreleased
assert "\n### " not in unreleased
+240
View File
@@ -0,0 +1,240 @@
# -*- coding: utf-8 -*-
"""Unit tests for P6 portfolio alert helpers."""
from __future__ import annotations
import json
import unittest
from datetime import date
from src.services.portfolio_alerts import (
PortfolioRiskAlert,
evaluate_portfolio_risk_alert,
expand_symbol_targets,
normalize_portfolio_alert_parameters,
)
class FakeRiskService:
def __init__(self, report):
self.report = report
self.calls = []
def get_risk_report(self, **kwargs):
self.calls.append(kwargs)
return self.report
class FakePortfolioService:
def __init__(self, *, snapshot=None, accounts=None):
self.snapshot = snapshot or {}
self.accounts = accounts or []
def get_portfolio_snapshot(self, **_kwargs):
return self.snapshot
def list_accounts(self, include_inactive=False):
return self.accounts
def _risk_rule(alert_type: str, *, target="1", parameters=None) -> PortfolioRiskAlert:
return PortfolioRiskAlert(
target_scope="portfolio_account",
target=target,
alert_type=alert_type,
parameters=parameters or {},
metadata={"persisted_rule_id": 7, "effective_target": f"account:{target}"},
)
def _risk_report():
return {
"as_of": "2026-05-20",
"account_id": 1,
"currency": "USD",
"thresholds": {
"concentration_alert_pct": 35.0,
"drawdown_alert_pct": 10.0,
"stop_loss_alert_pct": 10.0,
"stop_loss_near_ratio": 0.8,
},
"stop_loss": {
"near_alert": True,
"triggered_count": 1,
"near_count": 2,
"items": [
{"account_id": 1, "symbol": "AAPL", "loss_pct": 12.0, "is_triggered": True},
{"account_id": 1, "symbol": "MSFT", "loss_pct": 8.5, "is_triggered": False},
],
},
"concentration": {
"top_weight_pct": 42.5,
"alert": True,
"total_market_value": 10000.0,
"top_positions": [{"symbol": "AAPL", "weight_pct": 42.5}],
},
"drawdown": {
"series_points": 5,
"current_drawdown_pct": 5.0,
"max_drawdown_pct": 20.0,
"alert": True,
"fx_stale": True,
},
}
class PortfolioAlertsTestCase(unittest.TestCase):
def test_normalizes_stop_loss_mode(self) -> None:
self.assertEqual(normalize_portfolio_alert_parameters("portfolio_stop_loss", {}), {"mode": "near"})
self.assertEqual(
normalize_portfolio_alert_parameters("portfolio_stop_loss", {"mode": "breach"}),
{"mode": "breach"},
)
with self.assertRaisesRegex(ValueError, "near or breach"):
normalize_portfolio_alert_parameters("portfolio_stop_loss", {"mode": "bad"})
def test_stop_loss_near_and_breach_are_account_level_triggers(self) -> None:
report = _risk_report()
risk_service = FakeRiskService(report)
near = evaluate_portfolio_risk_alert(
_risk_rule("portfolio_stop_loss", parameters={"mode": "near"}),
risk_service=risk_service,
)
breach = evaluate_portfolio_risk_alert(
_risk_rule("portfolio_stop_loss", parameters={"mode": "breach"}),
risk_service=FakeRiskService(report),
)
self.assertTrue(near["triggered"])
self.assertEqual(near["observed_value"], 12.0)
self.assertIn("2 affected symbols", near["message"])
self.assertTrue(breach["triggered"])
self.assertIn("1 affected symbols", breach["message"])
diagnostics = json.loads(near["diagnostics"])
self.assertEqual(diagnostics["account_id"], 1)
self.assertEqual(diagnostics["currency"], "USD")
self.assertEqual(diagnostics["as_of"], "2026-05-20")
self.assertEqual(diagnostics["top_affected_symbols"], ["AAPL", "MSFT"])
def test_concentration_uses_top_weight_pct(self) -> None:
result = evaluate_portfolio_risk_alert(
_risk_rule("portfolio_concentration"),
risk_service=FakeRiskService(_risk_report()),
)
self.assertTrue(result["triggered"])
self.assertEqual(result["observed_value"], 42.5)
self.assertEqual(result["threshold"], 35.0)
def test_drawdown_uses_risk_report_alert_and_max_drawdown(self) -> None:
result = evaluate_portfolio_risk_alert(
_risk_rule("portfolio_drawdown"),
risk_service=FakeRiskService(_risk_report()),
)
self.assertTrue(result["triggered"])
self.assertEqual(result["observed_value"], 20.0)
diagnostics = json.loads(result["diagnostics"])
self.assertEqual(diagnostics["current_drawdown_pct"], 5.0)
self.assertEqual(diagnostics["max_drawdown_pct"], 20.0)
self.assertTrue(diagnostics["fx_stale"])
def test_price_stale_triggers_on_stale_or_missing_position_price(self) -> None:
snapshot = {
"as_of": "2026-05-20",
"currency": "CNY",
"fx_stale": False,
"accounts": [
{
"account_id": 3,
"positions": [
{"symbol": "600519", "price_stale": True, "price_available": True},
{"symbol": "000001", "price_stale": True, "price_available": False},
],
}
],
}
result = evaluate_portfolio_risk_alert(
_risk_rule("portfolio_price_stale", target="3"),
portfolio_service=FakePortfolioService(snapshot=snapshot),
)
self.assertTrue(result["triggered"])
self.assertEqual(result["observed_value"], 2.0)
diagnostics = json.loads(result["diagnostics"])
self.assertEqual(diagnostics["account_id"], 3)
self.assertTrue(diagnostics["price_stale"])
self.assertFalse(diagnostics["data_available"])
self.assertEqual(diagnostics["top_affected_symbols"], ["600519", "000001"])
def test_portfolio_holdings_expansion_deduplicates_symbols_and_caps(self) -> None:
snapshot = {
"accounts": [
{"positions": [{"symbol": "aapl", "quantity": 2}, {"symbol": "AAPL", "quantity": 1}]},
{"positions": [{"symbol": "hk00700", "quantity": 3}, {"symbol": "ZERO", "quantity": 0}]},
]
}
targets, overflow = expand_symbol_targets(
target_scope="portfolio_holdings",
target="all",
config=None,
portfolio_service=FakePortfolioService(snapshot=snapshot),
)
self.assertEqual([item.symbol for item in targets], ["AAPL", "HK00700"])
self.assertEqual(overflow, 0)
def test_portfolio_holdings_expansion_preserves_exchange_identity_and_dedupes_equivalent_formats(self) -> None:
snapshot = {
"accounts": [
{
"positions": [
{"symbol": "SH000001", "quantity": 2},
{"symbol": "000001.SH", "quantity": 2},
{"symbol": "SZ000001", "quantity": 1},
{"symbol": "000001.SZ", "quantity": 1},
{"symbol": "000001", "quantity": 3},
{"symbol": "600519.SH", "quantity": 4},
{"symbol": "SH600519", "quantity": 4},
]
},
]
}
targets, overflow = expand_symbol_targets(
target_scope="portfolio_holdings",
target="all",
config=None,
portfolio_service=FakePortfolioService(snapshot=snapshot),
)
self.assertEqual([item.symbol for item in targets], ["SH000001", "SZ000001", "000001", "SH600519"])
self.assertEqual(overflow, 0)
def test_watchlist_expansion_refreshes_stock_list(self) -> None:
class Config:
stock_list = ["600519", "600519", "aapl"]
def __init__(self):
self.refreshed = False
def refresh_stock_list(self):
self.refreshed = True
self.stock_list = ["000001", "000001", "hk00700"]
config = Config()
targets, overflow = expand_symbol_targets(
target_scope="watchlist",
target="default",
config=config,
)
self.assertTrue(config.refreshed)
self.assertEqual([item.symbol for item in targets], ["000001", "HK00700"])
self.assertEqual(overflow, 0)
if __name__ == "__main__":
unittest.main()