211 lines
9.7 KiB
Python
211 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Convert a VTB broker XLSX report to the import JSON formats.
|
||
|
||
Only the cash-movement section is sent to the existing statement importer.
|
||
Securities are written to a sidecar file for the future portfolio importer.
|
||
"""
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import sys
|
||
import uuid
|
||
from collections import defaultdict
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from pathlib import Path
|
||
from zipfile import ZipFile
|
||
from xml.etree import ElementTree as ET
|
||
|
||
NS = {'m': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'}
|
||
EPOCH = datetime(1899, 12, 30)
|
||
COLS = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
|
||
OPERATION_NAMESPACE = uuid.UUID('7a4a1c7d-7d3c-4f53-9f45-5e7dbb2c6f31')
|
||
|
||
|
||
def excel_date(value):
|
||
if value in (None, ''):
|
||
return None
|
||
number = float(value)
|
||
date = EPOCH + timedelta(days=number)
|
||
return date.isoformat(timespec='seconds') + '+03:00'
|
||
|
||
|
||
def excel_day(value):
|
||
return excel_date(value)[:10] if excel_date(value) else None
|
||
|
||
|
||
def kopecks(value):
|
||
if value in (None, ''):
|
||
return 0
|
||
return int((Decimal(str(value)).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)) * 100)
|
||
|
||
|
||
def text(value):
|
||
return re.sub(r'\s+', ' ', str(value or '')).strip()
|
||
|
||
|
||
def read_rows(path):
|
||
with ZipFile(path) as book:
|
||
shared_root = ET.fromstring(book.read('xl/sharedStrings.xml'))
|
||
shared = [text(''.join(node.itertext())) for node in shared_root.findall('m:si', NS)]
|
||
root = ET.fromstring(book.read('xl/worksheets/sheet1.xml'))
|
||
rows = []
|
||
for row in root.findall('.//m:sheetData/m:row', NS):
|
||
cells = {}
|
||
for cell in row.findall('m:c', NS):
|
||
ref = cell.attrib.get('r', '')
|
||
col = re.match(r'[A-Z]+', ref).group(0)
|
||
value = cell.find('m:v', NS)
|
||
raw = value.text if value is not None else ''
|
||
if cell.attrib.get('t') == 's' and raw:
|
||
raw = shared[int(raw)]
|
||
cells[col] = raw
|
||
rows.append((int(row.attrib['r']), cells))
|
||
return rows
|
||
|
||
|
||
def row_text(cells):
|
||
return text(' '.join(str(value) for value in cells.values()))
|
||
|
||
|
||
def find_row(rows, phrase, start=0):
|
||
for index in range(start, len(rows)):
|
||
if phrase.lower() in row_text(rows[index][1]).lower():
|
||
return index
|
||
raise ValueError(f'Не найден раздел: {phrase}')
|
||
|
||
|
||
def metadata(rows):
|
||
account = None
|
||
period = re.search(r'период с (\d{2}\.\d{2}\.\d{4}) по (\d{2}\.\d{2}\.\d{4})', row_text(dict(rows)))
|
||
report_date = None
|
||
for _, cells in rows[:35]:
|
||
joined = row_text(cells)
|
||
match = re.search(r'(\d{20})\s*\(RUR\)', joined)
|
||
account = account or (match.group(1) if match else None)
|
||
if 'Дата формирования отчета' in joined:
|
||
for value in cells.values():
|
||
if value and re.fullmatch(r'\d+(?:\.\d+)?', str(value)):
|
||
report_date = excel_day(value)
|
||
if not account or not period:
|
||
raise ValueError('Не удалось определить счёт или период отчёта')
|
||
return account, period.groups(), report_date
|
||
|
||
|
||
def cash_transactions(rows, start, end):
|
||
transactions = []
|
||
occurrences = defaultdict(int)
|
||
for number, cells in rows[start + 1:end]:
|
||
if not cells.get('B') or not cells.get('C') or not cells.get('J'):
|
||
continue
|
||
try:
|
||
operation_at = excel_date(cells['B'])
|
||
amount = kopecks(cells['C'])
|
||
except (TypeError, ValueError):
|
||
continue
|
||
operation = text(cells.get('J'))
|
||
comment = text(cells.get('P'))
|
||
description = text(f'{operation}. {comment}'.strip('. '))
|
||
digest = hashlib.sha256(f'{operation_at}|{amount}|{description}'.encode()).hexdigest()[:16]
|
||
occurrences[digest] += 1
|
||
source_id = f'vtb-broker-cash:{digest}:{occurrences[digest]}'
|
||
transactions.append({
|
||
'operationAt': operation_at,
|
||
'amountSigned': amount,
|
||
'commission': 0,
|
||
'description': description,
|
||
'sourceId': source_id,
|
||
'operationId': str(uuid.uuid5(OPERATION_NAMESPACE, source_id)),
|
||
'_sourceRow': number,
|
||
})
|
||
if not transactions:
|
||
raise ValueError('Операции движения денежных средств не найдены')
|
||
return transactions
|
||
|
||
|
||
def portfolio(rows, holdings_start, movement_start, trades_start, trades_end, account, period, report_date):
|
||
positions = []
|
||
for number, cells in rows[holdings_start + 1:movement_start]:
|
||
instrument = text(cells.get('B'))
|
||
if not instrument or instrument.lower().startswith('итого') or not re.search(r'RU[A-Z0-9]{10}', instrument):
|
||
continue
|
||
positions.append({
|
||
'sourceRow': number,
|
||
'instrument': instrument,
|
||
'isin': next((part for part in instrument.split(', ') if re.fullmatch(r'RU[A-Z0-9]{10}', part)), None),
|
||
'quantity': cells.get('L') or cells.get('M') or cells.get('I') or cells.get('J'),
|
||
'price': cells.get('P'),
|
||
'valuation': cells.get('AF') or cells.get('AJ'),
|
||
})
|
||
trades = []
|
||
columns = {'instrument': 'B', 'concludedAt': 'C', 'side': 'F', 'quantity': 'H',
|
||
'priceCurrency': 'I', 'price': 'J', 'settlementCurrency': 'L',
|
||
'settlementAmount': 'M', 'nkd': 'O', 'settlementCommission': 'P',
|
||
'tradeCommission': 'R', 'plannedDeliveryDate': 'S', 'plannedPaymentDate': 'T',
|
||
'orderId': 'W', 'tradeId': 'Z', 'organizerTradeId': 'AC', 'venue': 'AK',
|
||
'comment': 'AN'}
|
||
for number, cells in rows[trades_start + 1:trades_end]:
|
||
if not cells.get('B') or not cells.get('C') or not cells.get('F'):
|
||
continue
|
||
try:
|
||
concluded_at = excel_date(cells['C'])
|
||
except (TypeError, ValueError):
|
||
continue
|
||
item = {key: text(cells.get(col)) for key, col in columns.items()}
|
||
item['sourceRow'] = number
|
||
item['concludedAt'] = concluded_at
|
||
item['sourceId'] = f"vtb-broker-trade:{item['tradeId'] or number}"
|
||
item['operationId'] = str(uuid.uuid5(OPERATION_NAMESPACE, item['sourceId']))
|
||
trades.append(item)
|
||
return {
|
||
'schemaVersion': 'broker-portfolio-1.0',
|
||
'bank': 'VTB_BROKER',
|
||
'accountNumber': account,
|
||
'reportPeriod': {'from': datetime.strptime(period[0], '%d.%m.%Y').date().isoformat(),
|
||
'to': datetime.strptime(period[1], '%d.%m.%Y').date().isoformat()},
|
||
'reportedAt': report_date + 'T00:00:00+03:00' if report_date else None,
|
||
'positions': positions,
|
||
'trades': trades,
|
||
}
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) not in (3, 4):
|
||
raise SystemExit('usage: convert_vtb_broker_xlsx.py INPUT_XLSX CASH_JSON [PORTFOLIO_JSON]')
|
||
input_path, cash_path = map(Path, sys.argv[1:3])
|
||
portfolio_path = Path(sys.argv[3]) if len(sys.argv) == 4 else None
|
||
rows = read_rows(input_path)
|
||
account, period, report_date = metadata(rows)
|
||
holdings_start = find_row(rows, 'Отчёт об остатках ценных бумаг')
|
||
movement_start = find_row(rows, 'Движение ценных бумаг', holdings_start)
|
||
cash_start = find_row(rows, 'Движение денежных средств')
|
||
trades_start = find_row(rows, 'Заключенные в отчетном периоде сделки с ценными бумагами')
|
||
trades_end = find_row(rows, 'Завершенные в отчетном периоде сделки с ценными бумагами', trades_start + 1)
|
||
transactions = cash_transactions(rows, cash_start, holdings_start)
|
||
assert len({item['sourceId'] for item in transactions}) == len(transactions), 'sourceId операции не уникальны'
|
||
opening = kopecks(rows[find_row(rows, 'Отчёт об остатках денежных средств') + 3][1].get('L'))
|
||
balance_row = rows[find_row(rows, 'Отчёт об остатках денежных средств') + 3][1]
|
||
closing = kopecks(balance_row.get('AF'))
|
||
cash = {'schemaVersion': '1.0', 'bank': 'VTB_BROKER',
|
||
'statement': {'accountNumber': account, 'currency': 'RUB',
|
||
'openingBalance': opening, 'closingBalance': closing,
|
||
'exportedAt': (report_date or transactions[-1]['operationAt'][:10]) + 'T00:00:00+03:00'},
|
||
'transactions': [{key: value for key, value in item.items() if not key.startswith('_')}
|
||
for item in transactions]}
|
||
calculated_closing = opening + sum(item['amountSigned'] for item in transactions)
|
||
if calculated_closing != closing:
|
||
raise ValueError(f'Баланс не сходится: операции дают {calculated_closing}, XLSX содержит {closing}')
|
||
cash_path.parent.mkdir(parents=True, exist_ok=True)
|
||
cash_path.write_text(json.dumps(cash, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
||
duplicate_candidates = len(transactions) - len({item['sourceId'].rsplit(':', 1)[0] for item in transactions})
|
||
if portfolio_path:
|
||
portfolio_path.parent.mkdir(parents=True, exist_ok=True)
|
||
portfolio_path.write_text(json.dumps(portfolio(rows, holdings_start, movement_start, trades_start, trades_end, account, period, report_date), ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
||
print(json.dumps({'transactions': len(transactions), 'cashNet': sum(item['amountSigned'] for item in transactions),
|
||
'openingBalance': opening, 'closingBalance': closing, 'duplicateCandidates': duplicate_candidates,
|
||
'cashJson': str(cash_path), 'portfolioJson': str(portfolio_path) if portfolio_path else None}, ensure_ascii=False))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|