102 lines
6.4 KiB
TypeScript
102 lines
6.4 KiB
TypeScript
import crypto from 'crypto';
|
|
import { pool } from '../db/pool';
|
|
import type { ImportPortfolioResponse, PortfolioFile, PortfolioTrade } from '@family-budget/shared';
|
|
|
|
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
|
|
|
function decimalValue(value: unknown, field: string, required = false): string | null {
|
|
if (value == null || value === '') {
|
|
if (required) throw new Error(`${field} is required`);
|
|
return null;
|
|
}
|
|
const result = String(value).replace(/\s/g, '').replace(',', '.');
|
|
if (!/^-?(?:\d+\.?\d*|\.\d+)$/.test(result)) throw new Error(`${field} must be numeric`);
|
|
return result;
|
|
}
|
|
|
|
export function deriveTradeSourceId(trade: PortfolioTrade): string {
|
|
if (trade.sourceId?.trim()) return trade.sourceId.trim();
|
|
return [trade.isin || trade.instrument, trade.concludedAt, trade.side, decimalValue(trade.quantity, 'quantity', true), decimalValue(trade.settlementAmount, 'settlementAmount') || ''].join('|');
|
|
}
|
|
|
|
export function validatePortfolio(body: unknown): asserts body is PortfolioFile {
|
|
if (!body || typeof body !== 'object') throw new Error('Portfolio file must be an object');
|
|
const data = body as PortfolioFile;
|
|
if (data.schemaVersion !== 'broker-portfolio-1.0' || typeof data.bank !== 'string' || !data.bank.trim() || typeof data.accountNumber !== 'string' || !data.accountNumber.trim()) throw new Error('Invalid portfolio file');
|
|
if (!data.reportPeriod?.from || !data.reportPeriod?.to || Number.isNaN(Date.parse(data.reportPeriod.from)) || Number.isNaN(Date.parse(data.reportPeriod.to)) || data.reportPeriod.from > data.reportPeriod.to) throw new Error('Invalid report period');
|
|
if (data.reportedAt !== null && (typeof data.reportedAt !== 'string' || Number.isNaN(Date.parse(data.reportedAt)))) throw new Error('Invalid reportedAt');
|
|
if (!Array.isArray(data.positions) || !Array.isArray(data.trades)) throw new Error('positions and trades must be arrays');
|
|
const sourceIds = new Set<string>();
|
|
for (const trade of data.trades) {
|
|
if (!trade || typeof trade.instrument !== 'string' || !trade.instrument.trim() || typeof trade.concludedAt !== 'string' || Number.isNaN(Date.parse(trade.concludedAt)) || typeof trade.side !== 'string' || !trade.side.trim()) throw new Error('Invalid trade');
|
|
const sourceId = deriveTradeSourceId(trade);
|
|
if (sourceIds.has(sourceId)) throw new Error(`Duplicate sourceId: ${sourceId}`);
|
|
sourceIds.add(sourceId);
|
|
if (trade.operationId && !UUID_RE.test(trade.operationId)) throw new Error(`Invalid operationId: ${trade.sourceId}`);
|
|
decimalValue(trade.quantity, 'quantity', true);
|
|
}
|
|
for (const position of data.positions) {
|
|
if (!position || typeof position.instrument !== 'string' || !position.instrument.trim()) throw new Error('Invalid position');
|
|
decimalValue(position.quantity, 'position.quantity', true);
|
|
decimalValue(position.price, 'position.price');
|
|
decimalValue(position.valuation, 'position.valuation');
|
|
}
|
|
}
|
|
|
|
export async function importPortfolio(body: unknown, db: Pick<typeof pool, 'connect'> = pool): Promise<ImportPortfolioResponse> {
|
|
validatePortfolio(body);
|
|
const data = body;
|
|
const sourceHash = crypto.createHash('sha256').update(JSON.stringify(data)).digest('hex');
|
|
const client = await db.connect();
|
|
try {
|
|
await client.query('BEGIN');
|
|
const accountResult = await client.query(
|
|
`INSERT INTO accounts (bank, account_number, currency, account_type)
|
|
VALUES ($1, $2, 'RUB', 'brokerage')
|
|
ON CONFLICT (bank, account_number) DO UPDATE SET account_type = COALESCE(accounts.account_type, 'brokerage')
|
|
RETURNING id`,
|
|
[data.bank, data.accountNumber],
|
|
);
|
|
const accountId = Number(accountResult.rows[0].id);
|
|
const reportResult = await client.query(
|
|
`INSERT INTO portfolio_reports (account_id, source_hash, report_period_from, report_period_to, reported_at)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (account_id, source_hash) DO NOTHING
|
|
RETURNING id`,
|
|
[accountId, sourceHash, data.reportPeriod.from, data.reportPeriod.to, data.reportedAt],
|
|
);
|
|
if (reportResult.rows.length === 0) {
|
|
const existing = await client.query('SELECT id FROM portfolio_reports WHERE account_id = $1 AND source_hash = $2', [accountId, sourceHash]);
|
|
await client.query('COMMIT');
|
|
return { accountId, reportId: Number(existing.rows[0].id), importedTrades: 0, duplicateTrades: data.trades.length, positions: 0 };
|
|
}
|
|
const reportId = Number(reportResult.rows[0].id);
|
|
for (const position of data.positions) {
|
|
await client.query(
|
|
`INSERT INTO portfolio_positions (report_id, instrument, isin, quantity, price, valuation)
|
|
VALUES ($1, $2, $3, $4, $5, $6)`,
|
|
[reportId, position.instrument, position.isin ?? null, decimalValue(position.quantity, 'position.quantity', true), decimalValue(position.price, 'position.price'), decimalValue(position.valuation, 'position.valuation')],
|
|
);
|
|
}
|
|
let importedTrades = 0;
|
|
for (const trade of data.trades) {
|
|
const sourceId = deriveTradeSourceId(trade);
|
|
const result = await client.query(
|
|
`INSERT INTO portfolio_trades
|
|
(account_id, source_id, operation_id, instrument, isin, concluded_at, side, quantity, price_currency, price, settlement_currency, settlement_amount, nkd, settlement_commission, trade_commission, order_id, trade_id, venue, comment)
|
|
VALUES ($1, $2, COALESCE($3::uuid, gen_random_uuid()), $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
|
|
ON CONFLICT (account_id, source_id) DO NOTHING`,
|
|
[accountId, sourceId, trade.operationId ?? null, trade.instrument, trade.isin ?? null, trade.concludedAt, trade.side, decimalValue(trade.quantity, 'quantity', true), trade.priceCurrency ?? null, decimalValue(trade.price, 'price'), trade.settlementCurrency ?? null, decimalValue(trade.settlementAmount, 'settlementAmount'), decimalValue(trade.nkd, 'nkd'), decimalValue(trade.settlementCommission, 'settlementCommission'), decimalValue(trade.tradeCommission, 'tradeCommission'), trade.orderId ?? null, trade.tradeId ?? null, trade.venue ?? null, trade.comment ?? null],
|
|
);
|
|
importedTrades += result.rowCount ?? 0;
|
|
}
|
|
await client.query('COMMIT');
|
|
return { accountId, reportId, importedTrades, duplicateTrades: data.trades.length - importedTrades, positions: data.positions.length };
|
|
} catch (error) {
|
|
await client.query('ROLLBACK');
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|