#!/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 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' 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 transactions.append({ 'operationAt': operation_at, 'amountSigned': amount, 'commission': 0, 'description': description, 'sourceId': f'vtb-broker-cash:{digest}:{occurrences[digest]}', '_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}" 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')) closing = sum(item['amountSigned'] for item in transactions) + opening 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]} if opening + sum(item['amountSigned'] for item in transactions) != closing: raise ValueError('Баланс не сходится') 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()