Разработка системы автоматического импорта транзакций с бирж для учета

Разработка системы автоматического импорта транзакций с бирж для учета

Ручной CSV экспорт с каждой биржи — неудобный процесс: пользователи забывают, биржи меняют форматы, данные теряются. Автоматический импорт через API решает это: транзакции синхронизируются в фоне без участия пользователя.

API интеграции основных бирж

Binance API

class BinanceImporter implements ExchangeImporter {
  private client: Binance;
  
  async importTransactions(apiKey: string, secretKey: string, since: Date): Promise<RawTransaction[]> {
    this.client = new Binance({ apiKey, secretKey });
    
    const results = await Promise.all([
      this.getSpotTrades(since),
      this.getConversions(since),
      this.getStakingHistory(since),
      this.getSavingsInterest(since),
      this.getFlexibleEarnings(since),
      this.getDustConversions(since),
    ]);
    
    return results.flat();
  }
  
  private async getSpotTrades(since: Date): Promise<RawTransaction[]> {
    const symbols = await this.client.exchangeInfo().then(info => 
      info.symbols.map(s => s.symbol)
    );
    
    const trades: RawTransaction[] = [];
    
    // Binance требует запрос по каждой паре
    for (const symbol of symbols) {
      const symbolTrades = await this.client.myTrades({
        symbol,
        startTime: since.getTime(),
        limit: 1000,
      });
      
      trades.push(...symbolTrades.map(t => this.normalizeBinanceTrade(t)));
      
      // Rate limiting
      await sleep(100);
    }
    
    return trades;
  }
  
  private normalizeBinanceTrade(trade: any): RawTransaction {
    const baseAsset = trade.symbol.replace(/(USDT|BTC|ETH|BNB)$/, "");
    const quoteAsset = trade.symbol.slice(baseAsset.length);
    
    return {
      id: trade.id.toString(),
      timestamp: new Date(trade.time),
      type: trade.isBuyer ? "buy" : "sell",
      assetIn: trade.isBuyer ? baseAsset : quoteAsset,
      amountIn: trade.isBuyer ? parseFloat(trade.qty) : parseFloat(trade.quoteQty),
      assetOut: trade.isBuyer ? quoteAsset : baseAsset,
      amountOut: trade.isBuyer ? parseFloat(trade.quoteQty) : parseFloat(trade.qty),
      fee: parseFloat(trade.commission),
      feeCurrency: trade.commissionAsset,
      exchange: "BINANCE",
      txId: trade.orderId.toString(),
    };
  }
}

Coinbase Advanced Trade API

class CoinbaseImporter implements ExchangeImporter {
  async importTransactions(apiKey: string, secret: string, since: Date): Promise<RawTransaction[]> {
    const results = await Promise.all([
      this.getFills(apiKey, secret, since),
      this.getConversions(apiKey, secret, since),
      this.getRewards(apiKey, secret, since), // staking rewards
    ]);
    
    return results.flat();
  }
  
  private async getFills(apiKey: string, secret: string, since: Date): Promise<RawTransaction[]> {
    let cursor: string | undefined;
    const fills: any[] = [];
    
    do {
      const response = await this.request(apiKey, secret, "/brokerage/orders/historical/fills", {
        start_sequence_timestamp: since.toISOString(),
        cursor,
      });
      
      fills.push(...response.fills);
      cursor = response.cursor;
    } while (cursor);
    
    return fills.map(this.normalizeCoinbaseFill);
  }
}

Refresh scheduler

@Injectable()
class ExchangeSyncScheduler {
  // Синхронизируем каждые 4 часа для активных пользователей
  @Cron("0 */4 * * *")
  async syncActiveUsers() {
    const users = await this.db.getUsersWithApiKeys({
      lastSyncBefore: new Date(Date.now() - 3 * 60 * 60 * 1000), // > 3 часов назад
      isActive: true,
    });
    
    // Используем queue для параллельной обработки без перегрузки
    for (const user of users) {
      await this.syncQueue.add("sync-exchange", {
        userId: user.id,
        since: user.lastSyncAt,
      }, {
        attempts: 3,
        backoff: { type: "exponential", delay: 5000 },
      });
    }
  }
}

class ExchangeSyncWorker {
  async processJob(job: Job<SyncJobData>) {
    const { userId, since } = job.data;
    const exchangeConnections = await this.db.getUserExchanges(userId);
    
    for (const connection of exchangeConnections) {
      try {
        const importer = this.importerFactory.create(connection.exchange);
        const transactions = await importer.importTransactions(
          connection.apiKey,
          connection.secretKey,
          since
        );
        
        const normalized = transactions.map(tx => this.normalizer.normalize(tx));
        const classified = await this.classifier.classifyBatch(normalized, userId);
        
        await this.db.upsertTransactions(userId, classified);
        await this.db.updateLastSync(userId, connection.exchange);
        
      } catch (err) {
        if (err instanceof ApiKeyExpiredError) {
          await this.notifyUserApiKeyExpired(userId, connection.exchange);
        }
        throw err;
      }
    }
  }
}

Поддерживаемые биржи

Система автоматического импорта с 6+ биржами, background sync и rate limit handling — 3-4 недели разработки.