Агрегация

README.md

@@ -2,16 +2,6 @@
 
 [Kaggle Bitcoin Historical Data](https://www.kaggle.com/datasets/mczielinski/bitcoin-historical-data)
 
-Исходные данные хранят информацию по каждой минуте. Чтобы увеличить объём данных
-для более наглядной демонстрации эффективности параллельных вычислений
-и вычислений на GPU, с помощью линейной интерполяции данные были преобразованы
-из данных о каждой минуте в данные о каждых 10 секундах, то есть объём данных увеличился
-в 6 раз.
-
-```
-python3 upsample.py -i ./data/data.csv -o ./data/data_10s.csv -s 10
-```
-
 ## Задание
 
 Группируем данные по дням (Timestamp), за каждый день вычисляем среднюю цену 
@@ -20,6 +10,26 @@ python3 upsample.py -i ./data/data.csv -o ./data/data_10s.csv -s 10
 не менее чем на 10% от даты начала интервала, вместе с минимальными и максимальными 
 значениями Open и Close за все дни внутри интервала.
 
+## Параллельное чтение данных
+
+Нет смысла параллельно читать данные из NFS, так как в реальности файлы с данными
+будут лежать только на NFS сервере. То есть другие узлы лишь отправляют сетевые запросы
+на NFS сервер, который уже читает реальные данные с диска и лишь затем отправляет
+их другим узлам.
+
+Чтобы этого избежать, нужно на всех машинах скопировать файлы с данными в их реальные
+файловые системы. Например в папку `/data`.
+
+```sh
+# На каждом узле создаем директорию /data
+sudo mkdir /data
+sudo chown $USER /data
+
+# Копируем данные
+cd /mnt/shared/supercomputers/data
+cp data.csv /data/
+```
+
 ## Сборка
 
 Проект обязательно должен быть расположен в общей директории для всех узлов,

benchmark.py

@@ -1,115 +0,0 @@
-"""
-Запускает make run <number_of_runs> раз и считает статистику по времени выполнения.
-Тупо парсит out.txt и берём значение из строки "Total execution time: <time> sec".
-
-python benchmark.py <number_of_runs>
-"""
-
-import os
-import re
-import sys
-import time
-import subprocess
-import statistics
-
-N = int(sys.argv[1]) if len(sys.argv) > 1 else 10
-OUT = "out.txt"
-
-TIME_RE = re.compile(r"Total execution time:\s*([0-9]*\.?[0-9]+)\s*sec")
-JOB_RE  = re.compile(r"Submitted batch job\s+(\d+)")
-
-APPEAR_TIMEOUT = 300.0     # ждать появления out.txt
-FINISH_TIMEOUT = 3600.0    # ждать появления Total execution time (сек)
-POLL = 0.2                 # частота проверки файла
-
-def wait_for_exists(path: str, timeout: float):
-    t0 = time.time()
-    while not os.path.exists(path):
-        if time.time() - t0 > timeout:
-            raise TimeoutError(f"{path} did not appear within {timeout} seconds")
-        time.sleep(POLL)
-
-def try_read(path: str) -> str:
-    try:
-        with open(path, "r", encoding="utf-8", errors="replace") as f:
-            return f.read()
-    except FileNotFoundError:
-        return ""
-    except OSError:
-        # бывает, что файл на NFS в момент записи недоступен на чтение
-        return ""
-
-def wait_for_time_line(path: str, timeout: float) -> float:
-    t0 = time.time()
-    last_report = 0.0
-    while True:
-        txt = try_read(path)
-        matches = TIME_RE.findall(txt)
-        if matches:
-            return float(matches[-1])  # последняя встреченная строка
-
-        now = time.time()
-        if now - t0 > timeout:
-            tail = txt[-800:] if txt else "<empty>"
-            raise TimeoutError("Timed out waiting for 'Total execution time' line.\n"
-                               f"Last 800 chars of out.txt:\n{tail}")
-
-        # иногда полезно печатать прогресс раз в ~5 сек
-        if now - last_report > 5.0:
-            last_report = now
-            if txt:
-                # показать последнюю непустую строку
-                lines = [l for l in txt.splitlines() if l.strip()]
-                if lines:
-                    print(f"  waiting... last line: {lines[-1][:120]}", flush=True)
-                else:
-                    print("  waiting... (out.txt empty)", flush=True)
-            else:
-                print("  waiting... (out.txt not readable yet)", flush=True)
-
-        time.sleep(POLL)
-
-times = []
-
-for i in range(N):
-    print(f"Run {i+1}/{N} ...", flush=True)
-
-    # удаляем out.txt перед запуском
-    try:
-        os.remove(OUT)
-    except FileNotFoundError:
-        pass
-
-    # запускаем make run и забираем stdout (там будет Submitted batch job XXX)
-    res = subprocess.run(["make", "run"], capture_output=True, text=True)
-    out = (res.stdout or "") + "\n" + (res.stderr or "")
-
-    job_id = None
-    m = JOB_RE.search(out)
-    if m:
-        job_id = m.group(1)
-        print(f"  submitted job {job_id}", flush=True)
-    else:
-        print("  (job id not detected; will only watch out.txt)", flush=True)
-
-    # ждём появления out.txt и появления строки с Total execution time
-    wait_for_exists(OUT, APPEAR_TIMEOUT)
-    t = wait_for_time_line(OUT, FINISH_TIMEOUT)
-
-    times.append(t)
-    print(f"  time = {t:.3f} sec", flush=True)
-
-    # опционально удалить out.txt после парсинга
-    try:
-        os.remove(OUT)
-    except FileNotFoundError:
-        pass
-
-print("\n=== RESULTS ===")
-print(f"Runs:   {len(times)}")
-print(f"Mean:   {statistics.mean(times):.3f} sec")
-print(f"Median: {statistics.median(times):.3f} sec")
-print(f"Min:    {min(times):.3f} sec")
-print(f"Max:    {max(times):.3f} sec")
-if len(times) > 1:
-    print(f"Stddev: {statistics.stdev(times):.3f} sec")

run.slurm

@@ -6,19 +6,13 @@
 #SBATCH --output=out.txt
 
 # Путь к файлу данных (должен существовать на всех узлах)
-export DATA_PATH="/mnt/shared/supercomputers/data/data_10s.csv"
+export DATA_PATH="/mnt/shared/supercomputers/data/data.csv"
 
 # Доли данных для каждого ранка (сумма определяет пропорции)
-export DATA_READ_SHARES="10,11,13,14"
+export DATA_READ_SHARES="10,12,13,13"
 
 # Размер перекрытия в байтах для обработки границ строк
 export READ_OVERLAP_BYTES=131072
 
-# Интервал агрегации в секундах (60 = минуты, 600 = 10 минут, 86400 = дни)
-export AGGREGATION_INTERVAL=60
-
-# Использовать ли CUDA для агрегации (0 = нет, 1 = да)
-export USE_CUDA=1
-
 cd /mnt/shared/supercomputers/build
 mpirun -np $SLURM_NTASKS ./bitcoin_app

src/aggregation.cpp

@@ -1,75 +1,48 @@
 #include "aggregation.hpp"
-#include "utils.hpp"
+#include <map>
-
 #include <algorithm>
-#include <cstdint>
 #include <limits>
-#include <vector>
 
-std::vector<PeriodStats> aggregate_periods(const std::vector<Record>& records) {
+std::vector<DayStats> aggregate_days(const std::vector<Record>& records) {
-    const int64_t interval = get_aggregation_interval();
+    // Накопители для каждого дня
-
+    struct DayAccumulator {
-    std::vector<PeriodStats> result;
-    if (records.empty()) return result;
-
-    struct PeriodAccumulator {
         double avg_sum = 0.0;
         double open_min = std::numeric_limits<double>::max();
         double open_max = std::numeric_limits<double>::lowest();
         double close_min = std::numeric_limits<double>::max();
         double close_max = std::numeric_limits<double>::lowest();
         int64_t count = 0;
-
-        void add(const Record& r) {
-            const double avg = (r.low + r.high) / 2.0;
-            avg_sum += avg;
-            open_min = std::min(open_min, r.open);
-            open_max = std::max(open_max, r.open);
-            close_min = std::min(close_min, r.close);
-            close_max = std::max(close_max, r.close);
-            ++count;
-        }
     };
     
-    PeriodIndex current_period =
+    std::map<DayIndex, DayAccumulator> days;
-        static_cast<PeriodIndex>(records[0].timestamp) / interval;
     
-    PeriodAccumulator acc;
+    for (const auto& r : records) {
-    acc.add(records[0]);
+        DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
+        auto& acc = days[day];
         
-    for (size_t i = 1; i < records.size(); ++i) {
+        double avg = (r.low + r.high) / 2.0;
-        const Record& r = records[i];
+        acc.avg_sum += avg;
-        const PeriodIndex period =
+        acc.open_min = std::min(acc.open_min, r.open);
-            static_cast<PeriodIndex>(r.timestamp) / interval;
+        acc.open_max = std::max(acc.open_max, r.open);
-
+        acc.close_min = std::min(acc.close_min, r.close);
-        if (period != current_period) {
+        acc.close_max = std::max(acc.close_max, r.close);
-            PeriodStats stats;
+        acc.count++;
-            stats.period = current_period;
-            stats.avg = acc.avg_sum / static_cast<double>(acc.count);
-            stats.open_min = acc.open_min;
-            stats.open_max = acc.open_max;
-            stats.close_min = acc.close_min;
-            stats.close_max = acc.close_max;
-            stats.count = acc.count;
-            result.push_back(stats);
-
-            current_period = period;
-            acc = PeriodAccumulator{};
-        }
-
-        acc.add(r);
     }
     
-    // последний период
+    std::vector<DayStats> result;
-    PeriodStats stats;
+    result.reserve(days.size());
-    stats.period = current_period;
+    
-    stats.avg = acc.avg_sum / static_cast<double>(acc.count);
+    for (const auto& [day, acc] : days) {
-    stats.open_min = acc.open_min;
+        DayStats stats;
-    stats.open_max = acc.open_max;
+        stats.day = day;
-    stats.close_min = acc.close_min;
+        stats.avg = acc.avg_sum / static_cast<double>(acc.count);
-    stats.close_max = acc.close_max;
+        stats.open_min = acc.open_min;
-    stats.count = acc.count;
+        stats.open_max = acc.open_max;
-    result.push_back(stats);
+        stats.close_min = acc.close_min;
+        stats.close_max = acc.close_max;
+        stats.count = acc.count;
+        result.push_back(stats);
+    }
     
     return result;
 }

src/aggregation.hpp

@@ -1,8 +1,8 @@
 #pragma once
 
 #include "record.hpp"
-#include "period_stats.hpp"
+#include "day_stats.hpp"
 #include <vector>
 
-// Агрегация записей по периодам на одном узле
+// Агрегация записей по дням на одном узле
-std::vector<PeriodStats> aggregate_periods(const std::vector<Record>& records);
+std::vector<DayStats> aggregate_days(const std::vector<Record>& records);

src/day_stats.hpp

@@ -0,0 +1,15 @@
+#pragma once
+#include <cstdint>
+
+using DayIndex = int64_t;
+
+// Агрегированные данные за один день
+struct DayStats {
+    DayIndex day;       // индекс дня (timestamp / 86400)
+    double avg;         // среднее значение (Low + High) / 2 по всем записям
+    double open_min;    // минимальный Open за день
+    double open_max;    // максимальный Open за день
+    double close_min;   // минимальный Close за день
+    double close_max;   // максимальный Close за день
+    int64_t count;      // количество записей, по которым агрегировали
+};

src/gpu_loader.cpp

@@ -1,133 +1,145 @@
 #include "gpu_loader.hpp"
 #include <dlfcn.h>
+#include <map>
+#include <algorithm>
 #include <iostream>
-#include <cstdint>
+#include <iomanip>
-
+#include <omp.h>
-// Структура результата GPU (должна совпадать с gpu_plugin.cu)
-struct GpuPeriodStats {
-    int64_t period;
-    double avg;
-    double open_min;
-    double open_max;
-    double close_min;
-    double close_max;
-    int64_t count;
-};
-
-// Типы функций из GPU плагина
-using gpu_is_available_fn = int (*)();
-
-using gpu_aggregate_periods_fn = int (*)(
-    const double* h_timestamps,
-    const double* h_open,
-    const double* h_high,
-    const double* h_low,
-    const double* h_close,
-    int num_ticks,
-    int64_t interval,
-    GpuPeriodStats** h_out_stats,
-    int* out_num_periods
-);
-
-using gpu_free_results_fn = void (*)(GpuPeriodStats*);
 
 static void* get_gpu_lib_handle() {
     static void* h = dlopen("./libgpu_compute.so", RTLD_NOW | RTLD_LOCAL);
     return h;
 }
 
-bool gpu_is_available() {
+gpu_is_available_fn load_gpu_is_available() {
     void* h = get_gpu_lib_handle();
-    if (!h) return false;
+    if (!h) return nullptr;
 
-    auto fn = reinterpret_cast<gpu_is_available_fn>(dlsym(h, "gpu_is_available"));
+    auto fn = (gpu_is_available_fn)dlsym(h, "gpu_is_available");
-    if (!fn) return false;
+    return fn;
-    
-    return fn() != 0;
 }
 
-bool aggregate_periods_gpu(
+bool gpu_is_available() {    
-    const std::vector<Record>& records,
+    auto gpu_is_available_fn = load_gpu_is_available();
-    int64_t aggregation_interval,
+    
-    std::vector<PeriodStats>& out_stats)
+    if (gpu_is_available_fn && gpu_is_available_fn()) {
-{
-    if (records.empty()) {
-        out_stats.clear();
         return true;
     }
     
+    return false;
+}
+
+gpu_aggregate_days_fn load_gpu_aggregate_days() {
     void* h = get_gpu_lib_handle();
-    if (!h) {
+    if (!h) return nullptr;
-        std::cerr << "GPU: Failed to load libgpu_compute.so" << std::endl;
+
+    auto fn = (gpu_aggregate_days_fn)dlsym(h, "gpu_aggregate_days");
+    return fn;
+}
+
+bool aggregate_days_gpu(
+    const std::vector<Record>& records,
+    std::vector<DayStats>& out_stats,
+    gpu_aggregate_days_fn gpu_fn)
+{
+    if (!gpu_fn || records.empty()) {
         return false;
     }
 
-    auto aggregate_fn = reinterpret_cast<gpu_aggregate_periods_fn>(
+    // Общий таймер всей функции
-        dlsym(h, "gpu_aggregate_periods"));
+    double t_total_start = omp_get_wtime();
-    auto free_fn = reinterpret_cast<gpu_free_results_fn>(
-        dlsym(h, "gpu_free_results"));
     
-    if (!aggregate_fn || !free_fn) {
+    // Таймер CPU preprocessing
-        std::cerr << "GPU: Failed to load functions from plugin" << std::endl;
+    double t_preprocess_start = omp_get_wtime();
-        return false;
+
+    // Группируем записи по дням и подготавливаем данные для GPU
+    std::map<DayIndex, std::vector<size_t>> day_record_indices;
+    
+    for (size_t i = 0; i < records.size(); i++) {
+        DayIndex day = static_cast<DayIndex>(records[i].timestamp) / 86400;
+        day_record_indices[day].push_back(i);
     }
 
-    int num_ticks = static_cast<int>(records.size());
+    int num_days = static_cast<int>(day_record_indices.size());
     
-    // Конвертируем AoS в SoA
+    // Подготавливаем массивы для GPU
-    std::vector<double> timestamps(num_ticks);
+    std::vector<GpuRecord> gpu_records;
-    std::vector<double> open(num_ticks);
+    std::vector<int> day_offsets;
-    std::vector<double> high(num_ticks);
+    std::vector<int> day_counts;
-    std::vector<double> low(num_ticks);
+    std::vector<long long> day_indices;
-    std::vector<double> close(num_ticks);
     
-    for (int i = 0; i < num_ticks; i++) {
+    gpu_records.reserve(records.size());
-        timestamps[i] = records[i].timestamp;
+    day_offsets.reserve(num_days);
-        open[i] = records[i].open;
+    day_counts.reserve(num_days);
-        high[i] = records[i].high;
+    day_indices.reserve(num_days);
-        low[i] = records[i].low;
+    
-        close[i] = records[i].close;
+    int current_offset = 0;
+    
+    for (auto& [day, indices] : day_record_indices) {
+        day_indices.push_back(day);
+        day_offsets.push_back(current_offset);
+        day_counts.push_back(static_cast<int>(indices.size()));
+        
+        // Добавляем записи этого дня
+        for (size_t idx : indices) {
+            const auto& r = records[idx];
+            GpuRecord gr;
+            gr.timestamp = r.timestamp;
+            gr.open = r.open;
+            gr.high = r.high;
+            gr.low = r.low;
+            gr.close = r.close;
+            gr.volume = r.volume;
+            gpu_records.push_back(gr);
+        }
+        
+        current_offset += static_cast<int>(indices.size());
     }
     
-    // Вызываем GPU функцию
+    // Выделяем память для результата
-    GpuPeriodStats* gpu_stats = nullptr;
+    std::vector<GpuDayStats> gpu_stats(num_days);
-    int num_periods = 0;
     
-    int result = aggregate_fn(
+    double t_preprocess_ms = (omp_get_wtime() - t_preprocess_start) * 1000.0;
-        timestamps.data(),
+    std::cout << "  GPU CPU preprocessing:  " << std::fixed << std::setprecision(3) 
-        open.data(),
+              << std::setw(7) << t_preprocess_ms << " ms" << std::endl << std::flush;
-        high.data(),
+    
-        low.data(),
+    // Вызываем GPU функцию (включает: malloc, memcpy H->D, kernel, memcpy D->H, free)
-        close.data(),
+    // Детальные тайминги выводятся внутри GPU функции
-        num_ticks,
+    int result = gpu_fn(
-        aggregation_interval,
+        gpu_records.data(),
-        &gpu_stats,
+        static_cast<int>(gpu_records.size()),
-        &num_periods
+        day_offsets.data(),
+        day_counts.data(),
+        day_indices.data(),
+        num_days,
+        gpu_stats.data()
     );
     
     if (result != 0) {
-        std::cerr << "GPU: Aggregation failed with code " << result << std::endl;
+        std::cout << "  GPU: Function returned error code " << result << std::endl;
         return false;
     }
     
-    // Конвертируем результат в PeriodStats
+    // Конвертируем результат в DayStats
     out_stats.clear();
-    out_stats.reserve(num_periods);
+    out_stats.reserve(num_days);
     
-    for (int i = 0; i < num_periods; i++) {
+    for (const auto& gs : gpu_stats) {
-        PeriodStats ps;
+        DayStats ds;
-        ps.period = gpu_stats[i].period;
+        ds.day = gs.day;
-        ps.avg = gpu_stats[i].avg;
+        ds.avg = gs.avg;
-        ps.open_min = gpu_stats[i].open_min;
+        ds.open_min = gs.open_min;
-        ps.open_max = gpu_stats[i].open_max;
+        ds.open_max = gs.open_max;
-        ps.close_min = gpu_stats[i].close_min;
+        ds.close_min = gs.close_min;
-        ps.close_max = gpu_stats[i].close_max;
+        ds.close_max = gs.close_max;
-        ps.count = gpu_stats[i].count;
+        ds.count = gs.count;
-        out_stats.push_back(ps);
+        out_stats.push_back(ds);
     }
     
-    // Освобождаем память
+    // Общее время всей GPU функции (включая preprocessing)
-    free_fn(gpu_stats);
+    double t_total_ms = (omp_get_wtime() - t_total_start) * 1000.0;
+    std::cout << "  GPU TOTAL (with prep):  " << std::fixed << std::setprecision(3) 
+              << std::setw(7) << t_total_ms << " ms" << std::endl << std::flush;
     
     return true;
 }

src/gpu_loader.hpp

@@ -1,15 +1,50 @@
 #pragma once
-#include "period_stats.hpp"
+#include "day_stats.hpp"
 #include "record.hpp"
 #include <vector>
 
-// Проверка доступности CUDA
 bool gpu_is_available();
 
-// Агрегация периодов на GPU
+// Типы функций из GPU плагина
-// Возвращает true если успешно, false если GPU недоступен или ошибка
+using gpu_is_available_fn = int (*)();
-bool aggregate_periods_gpu(
+
-    const std::vector<Record>& records,
+// Структуры для GPU (должны совпадать с gpu_plugin.cu)
-    int64_t aggregation_interval,
+struct GpuRecord {
-    std::vector<PeriodStats>& out_stats
+    double timestamp;
+    double open;
+    double high;
+    double low;
+    double close;
+    double volume;
+};
+
+struct GpuDayStats {
+    long long day;
+    double avg;
+    double open_min;
+    double open_max;
+    double close_min;
+    double close_max;
+    long long count;
+};
+
+using gpu_aggregate_days_fn = int (*)(
+    const GpuRecord* h_records,
+    int num_records,
+    const int* h_day_offsets,
+    const int* h_day_counts,
+    const long long* h_day_indices,
+    int num_days,
+    GpuDayStats* h_out_stats
+);
+
+// Загрузка функций из плагина
+gpu_is_available_fn load_gpu_is_available();
+gpu_aggregate_days_fn load_gpu_aggregate_days();
+
+// Обёртка для агрегации на GPU (возвращает true если успешно)
+bool aggregate_days_gpu(
+    const std::vector<Record>& records,
+    std::vector<DayStats>& out_stats,
+    gpu_aggregate_days_fn gpu_fn
 );

src/gpu_plugin.cu

@@ -1,430 +1,262 @@
 #include <cuda_runtime.h>
-#include <cub/cub.cuh>
 #include <cstdint>
 #include <cfloat>
 #include <cstdio>
 #include <ctime>
-#include <string>
-#include <sstream>
-#include <iomanip>
-
-// ============================================================================
-// Структуры данных
-// ============================================================================
-
-// SoA (Structure of Arrays) для входных данных на GPU
-struct GpuTicksSoA {
-    double* timestamp;
-    double* open;
-    double* high;
-    double* low;
-    double* close;
-    int n;
-};
-
-// Результат агрегации одного периода
-struct GpuPeriodStats {
-    int64_t period;
-    double avg;
-    double open_min;
-    double open_max;
-    double close_min;
-    double close_max;
-    int64_t count;
-};
-
-// ============================================================================
-// Вспомогательные функции
-// ============================================================================
 
+// CPU таймер в миллисекундах
 static double get_time_ms() {
     struct timespec ts;
     clock_gettime(CLOCK_MONOTONIC, &ts);
     return ts.tv_sec * 1000.0 + ts.tv_nsec / 1000000.0;
 }
 
-#define CUDA_CHECK(call) do { \
+// Структуры данных (должны совпадать с C++ кодом)
-    cudaError_t err = call; \
+struct GpuRecord {
-    if (err != cudaSuccess) { \
+    double timestamp;
-        printf("CUDA error at %s:%d: %s\n", __FILE__, __LINE__, cudaGetErrorString(err)); \
+    double open;
-        return -1; \
+    double high;
-    } \
+    double low;
-} while(0)
+    double close;
+    double volume;
+};
 
-// ============================================================================
+struct GpuDayStats {
-// Kernel: вычисление period_id для каждого тика
+    long long day;
-// ============================================================================
+    double avg;
-
+    double open_min;
-__global__ void compute_period_ids_kernel(
+    double open_max;
-    const double* __restrict__ timestamps,
+    double close_min;
-    int64_t* __restrict__ period_ids,
+    double close_max;
-    int n,
+    long long count;
-    int64_t interval)
+};
-{
-    int idx = blockIdx.x * blockDim.x + threadIdx.x;
-    if (idx < n) {
-        period_ids[idx] = static_cast<int64_t>(timestamps[idx]) / interval;
-    }
-}
-
-// ============================================================================
-// Kernel: агрегация одного периода (один блок на период)
-// ============================================================================
-
-__global__ void aggregate_periods_kernel(
-    const double* __restrict__ open,
-    const double* __restrict__ high,
-    const double* __restrict__ low,
-    const double* __restrict__ close,
-    const int64_t* __restrict__ unique_periods,
-    const int* __restrict__ offsets,
-    const int* __restrict__ counts,
-    int num_periods,
-    GpuPeriodStats* __restrict__ out_stats)
-{
-    int period_idx = blockIdx.x;
-    if (period_idx >= num_periods) return;
-    
-    int offset = offsets[period_idx];
-    int count = counts[period_idx];
-    
-    // Используем shared memory для редукции внутри блока
-    __shared__ double s_avg_sum;
-    __shared__ double s_open_min;
-    __shared__ double s_open_max;
-    __shared__ double s_close_min;
-    __shared__ double s_close_max;
-    
-    // Инициализация shared memory первым потоком
-    if (threadIdx.x == 0) {
-        s_avg_sum = 0.0;
-        s_open_min = DBL_MAX;
-        s_open_max = -DBL_MAX;
-        s_close_min = DBL_MAX;
-        s_close_max = -DBL_MAX;
-    }
-    __syncthreads();
-    
-    // Локальные аккумуляторы для каждого потока
-    double local_avg_sum = 0.0;
-    double local_open_min = DBL_MAX;
-    double local_open_max = -DBL_MAX;
-    double local_close_min = DBL_MAX;
-    double local_close_max = -DBL_MAX;
-    
-    // Каждый поток обрабатывает свою часть тиков
-    for (int i = threadIdx.x; i < count; i += blockDim.x) {
-        int tick_idx = offset + i;
-        double avg = (low[tick_idx] + high[tick_idx]) / 2.0;
-        local_avg_sum += avg;
-        local_open_min = min(local_open_min, open[tick_idx]);
-        local_open_max = max(local_open_max, open[tick_idx]);
-        local_close_min = min(local_close_min, close[tick_idx]);
-        local_close_max = max(local_close_max, close[tick_idx]);
-    }
-    
-    // Редукция с использованием атомарных операций
-    atomicAdd(&s_avg_sum, local_avg_sum);
-    atomicMin(reinterpret_cast<unsigned long long*>(&s_open_min), 
-              __double_as_longlong(local_open_min));
-    atomicMax(reinterpret_cast<unsigned long long*>(&s_open_max),
-              __double_as_longlong(local_open_max));
-    atomicMin(reinterpret_cast<unsigned long long*>(&s_close_min),
-              __double_as_longlong(local_close_min));
-    atomicMax(reinterpret_cast<unsigned long long*>(&s_close_max),
-              __double_as_longlong(local_close_max));
-    
-    __syncthreads();
-    
-    // Первый поток записывает результат
-    if (threadIdx.x == 0) {
-        GpuPeriodStats stats;
-        stats.period = unique_periods[period_idx];
-        stats.avg = s_avg_sum / static_cast<double>(count);
-        stats.open_min = s_open_min;
-        stats.open_max = s_open_max;
-        stats.close_min = s_close_min;
-        stats.close_max = s_close_max;
-        stats.count = count;
-        out_stats[period_idx] = stats;
-    }
-}
-
-// ============================================================================
-// Простой kernel для агрегации (один поток на период)
-// Используется когда периодов много и тиков в каждом мало
-// ============================================================================
-
-__global__ void aggregate_periods_simple_kernel(
-    const double* __restrict__ open,
-    const double* __restrict__ high,
-    const double* __restrict__ low,
-    const double* __restrict__ close,
-    const int64_t* __restrict__ unique_periods,
-    const int* __restrict__ offsets,
-    const int* __restrict__ counts,
-    int num_periods,
-    GpuPeriodStats* __restrict__ out_stats)
-{
-    int period_idx = blockIdx.x * blockDim.x + threadIdx.x;
-    if (period_idx >= num_periods) return;
-    
-    int offset = offsets[period_idx];
-    int count = counts[period_idx];
-    
-    double avg_sum = 0.0;
-    double open_min = DBL_MAX;
-    double open_max = -DBL_MAX;
-    double close_min = DBL_MAX;
-    double close_max = -DBL_MAX;
-    
-    for (int i = 0; i < count; i++) {
-        int tick_idx = offset + i;
-        double avg = (low[tick_idx] + high[tick_idx]) / 2.0;
-        avg_sum += avg;
-        open_min = min(open_min, open[tick_idx]);
-        open_max = max(open_max, open[tick_idx]);
-        close_min = min(close_min, close[tick_idx]);
-        close_max = max(close_max, close[tick_idx]);
-    }
-    
-    GpuPeriodStats stats;
-    stats.period = unique_periods[period_idx];
-    stats.avg = avg_sum / static_cast<double>(count);
-    stats.open_min = open_min;
-    stats.open_max = open_max;
-    stats.close_min = close_min;
-    stats.close_max = close_max;
-    stats.count = count;
-    out_stats[period_idx] = stats;
-}
-
-// ============================================================================
-// Проверка доступности GPU
-// ============================================================================
 
 extern "C" int gpu_is_available() {
     int n = 0;
     cudaError_t err = cudaGetDeviceCount(&n);
     if (err != cudaSuccess) return 0;
     if (n > 0) {
-        cudaFree(0);  // Форсируем инициализацию контекста
+        // Инициализируем CUDA контекст заранее (cudaFree(0) форсирует инициализацию)
+        cudaFree(0);
     }
     return (n > 0) ? 1 : 0;
 }
 
-// ============================================================================
+// Kernel для агрегации (каждый поток обрабатывает один день)
-// Главная функция агрегации на GPU
+__global__ void aggregate_kernel(
-// ============================================================================
+    const GpuRecord* records,
-
+    int num_records,
-extern "C" int gpu_aggregate_periods(
+    const int* day_offsets,    // начало каждого дня в массиве records
-    const double* h_timestamps,
+    const int* day_counts,     // количество записей в каждом дне
-    const double* h_open,
+    const long long* day_indices, // индексы дней
-    const double* h_high,
+    int num_days,
-    const double* h_low,
+    GpuDayStats* out_stats)
-    const double* h_close,
-    int num_ticks,
-    int64_t interval,
-    GpuPeriodStats** h_out_stats,
-    int* out_num_periods)
 {
-    if (num_ticks == 0) {
+    // Глобальный индекс потока = индекс дня
-        *h_out_stats = nullptr;
+    int d = blockIdx.x * blockDim.x + threadIdx.x;
-        *out_num_periods = 0;
+    
-        return 0;
+    if (d >= num_days) return;
+    
+    int offset = day_offsets[d];
+    int count = day_counts[d];
+    
+    GpuDayStats stats;
+    stats.day = day_indices[d];
+    stats.open_min = DBL_MAX;
+    stats.open_max = -DBL_MAX;
+    stats.close_min = DBL_MAX;
+    stats.close_max = -DBL_MAX;
+    stats.count = count;
+    
+    double avg_sum = 0.0;
+    
+    for (int i = 0; i < count; i++) {
+        const GpuRecord& r = records[offset + i];
+        
+        // Accumulate avg = (low + high) / 2
+        avg_sum += (r.low + r.high) / 2.0;
+        
+        // min/max Open
+        if (r.open < stats.open_min) stats.open_min = r.open;
+        if (r.open > stats.open_max) stats.open_max = r.open;
+        
+        // min/max Close
+        if (r.close < stats.close_min) stats.close_min = r.close;
+        if (r.close > stats.close_max) stats.close_max = r.close;
     }
     
-    std::ostringstream output;
+    stats.avg = avg_sum / static_cast<double>(count);
-    double total_start = get_time_ms();
+    out_stats[d] = stats;
+}
 
-    // ========================================================================
+// Функция агрегации, вызываемая из C++
-    // Шаг 1: Выделение памяти и копирование данных на GPU
+extern "C" int gpu_aggregate_days(
-    // ========================================================================
+    const GpuRecord* h_records,
-    double step1_start = get_time_ms();
+    int num_records,
+    const int* h_day_offsets,
+    const int* h_day_counts,
+    const long long* h_day_indices,
+    int num_days,
+    GpuDayStats* h_out_stats)
+{
+    double cpu_total_start = get_time_ms();
     
-    double* d_timestamps = nullptr;
+    // === Создаём CUDA события для измерения времени ===
-    double* d_open = nullptr;
+    double cpu_event_create_start = get_time_ms();
-    double* d_high = nullptr;
-    double* d_low = nullptr;
-    double* d_close = nullptr;
-    int64_t* d_period_ids = nullptr;
     
-    size_t ticks_bytes = num_ticks * sizeof(double);
+    cudaEvent_t start_malloc, stop_malloc;
+    cudaEvent_t start_transfer, stop_transfer;
+    cudaEvent_t start_kernel, stop_kernel;
+    cudaEvent_t start_copy_back, stop_copy_back;
+    cudaEvent_t start_free, stop_free;
     
-    CUDA_CHECK(cudaMalloc(&d_timestamps, ticks_bytes));
+    cudaEventCreate(&start_malloc);
-    CUDA_CHECK(cudaMalloc(&d_open, ticks_bytes));
+    cudaEventCreate(&stop_malloc);
-    CUDA_CHECK(cudaMalloc(&d_high, ticks_bytes));
+    cudaEventCreate(&start_transfer);
-    CUDA_CHECK(cudaMalloc(&d_low, ticks_bytes));
+    cudaEventCreate(&stop_transfer);
-    CUDA_CHECK(cudaMalloc(&d_close, ticks_bytes));
+    cudaEventCreate(&start_kernel);
-    CUDA_CHECK(cudaMalloc(&d_period_ids, num_ticks * sizeof(int64_t)));
+    cudaEventCreate(&stop_kernel);
+    cudaEventCreate(&start_copy_back);
+    cudaEventCreate(&stop_copy_back);
+    cudaEventCreate(&start_free);
+    cudaEventCreate(&stop_free);
     
-    CUDA_CHECK(cudaMemcpy(d_timestamps, h_timestamps, ticks_bytes, cudaMemcpyHostToDevice));
+    double cpu_event_create_ms = get_time_ms() - cpu_event_create_start;
-    CUDA_CHECK(cudaMemcpy(d_open, h_open, ticks_bytes, cudaMemcpyHostToDevice));
-    CUDA_CHECK(cudaMemcpy(d_high, h_high, ticks_bytes, cudaMemcpyHostToDevice));
-    CUDA_CHECK(cudaMemcpy(d_low, h_low, ticks_bytes, cudaMemcpyHostToDevice));
-    CUDA_CHECK(cudaMemcpy(d_close, h_close, ticks_bytes, cudaMemcpyHostToDevice));
     
-    double step1_ms = get_time_ms() - step1_start;
+    // === ИЗМЕРЕНИЕ cudaMalloc ===
+    cudaEventRecord(start_malloc);
     
-    // ========================================================================
+    GpuRecord* d_records = nullptr;
-    // Шаг 2: Вычисление period_id для каждого тика
+    int* d_day_offsets = nullptr;
-    // ========================================================================
+    int* d_day_counts = nullptr;
-    double step2_start = get_time_ms();
+    long long* d_day_indices = nullptr;
+    GpuDayStats* d_out_stats = nullptr;
     
-    const int BLOCK_SIZE = 256;
+    cudaError_t err;
-    int num_blocks = (num_ticks + BLOCK_SIZE - 1) / BLOCK_SIZE;
     
-    compute_period_ids_kernel<<<num_blocks, BLOCK_SIZE>>>(
+    err = cudaMalloc(&d_records, num_records * sizeof(GpuRecord));
-        d_timestamps, d_period_ids, num_ticks, interval);
+    if (err != cudaSuccess) return -1;
-    CUDA_CHECK(cudaGetLastError());
-    CUDA_CHECK(cudaDeviceSynchronize());
     
-    double step2_ms = get_time_ms() - step2_start;
+    err = cudaMalloc(&d_day_offsets, num_days * sizeof(int));
+    if (err != cudaSuccess) { cudaFree(d_records); return -2; }
     
-    // ========================================================================
+    err = cudaMalloc(&d_day_counts, num_days * sizeof(int));
-    // Шаг 3: RLE (Run-Length Encode) для нахождения уникальных периодов
+    if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); return -3; }
-    // ========================================================================
-    double step3_start = get_time_ms();
     
-    int64_t* d_unique_periods = nullptr;
+    err = cudaMalloc(&d_day_indices, num_days * sizeof(long long));
-    int* d_counts = nullptr;
+    if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); cudaFree(d_day_counts); return -4; }
-    int* d_num_runs = nullptr;
     
-    CUDA_CHECK(cudaMalloc(&d_unique_periods, num_ticks * sizeof(int64_t)));
+    err = cudaMalloc(&d_out_stats, num_days * sizeof(GpuDayStats));
-    CUDA_CHECK(cudaMalloc(&d_counts, num_ticks * sizeof(int)));
+    if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); cudaFree(d_day_counts); cudaFree(d_day_indices); return -5; }
-    CUDA_CHECK(cudaMalloc(&d_num_runs, sizeof(int)));
     
-    // Определяем размер временного буфера для CUB
+    cudaEventRecord(stop_malloc);
-    void* d_temp_storage = nullptr;
+    cudaEventSynchronize(stop_malloc);
-    size_t temp_storage_bytes = 0;
     
-    cub::DeviceRunLengthEncode::Encode(
+    float time_malloc_ms = 0;
-        d_temp_storage, temp_storage_bytes,
+    cudaEventElapsedTime(&time_malloc_ms, start_malloc, stop_malloc);
-        d_period_ids, d_unique_periods, d_counts, d_num_runs,
-        num_ticks);
     
-    CUDA_CHECK(cudaMalloc(&d_temp_storage, temp_storage_bytes));
+    // === ИЗМЕРЕНИЕ memcpy H->D ===
+    cudaEventRecord(start_transfer);
     
-    cub::DeviceRunLengthEncode::Encode(
+    err = cudaMemcpy(d_records, h_records, num_records * sizeof(GpuRecord), cudaMemcpyHostToDevice);
-        d_temp_storage, temp_storage_bytes,
+    if (err != cudaSuccess) return -10;
-        d_period_ids, d_unique_periods, d_counts, d_num_runs,
-        num_ticks);
-    CUDA_CHECK(cudaGetLastError());
     
-    // Копируем количество уникальных периодов
+    err = cudaMemcpy(d_day_offsets, h_day_offsets, num_days * sizeof(int), cudaMemcpyHostToDevice);
-    int num_periods = 0;
+    if (err != cudaSuccess) return -11;
-    CUDA_CHECK(cudaMemcpy(&num_periods, d_num_runs, sizeof(int), cudaMemcpyDeviceToHost));
     
-    cudaFree(d_temp_storage);
+    err = cudaMemcpy(d_day_counts, h_day_counts, num_days * sizeof(int), cudaMemcpyHostToDevice);
-    d_temp_storage = nullptr;
+    if (err != cudaSuccess) return -12;
     
-    double step3_ms = get_time_ms() - step3_start;
+    err = cudaMemcpy(d_day_indices, h_day_indices, num_days * sizeof(long long), cudaMemcpyHostToDevice);
+    if (err != cudaSuccess) return -13;
     
-    // ========================================================================
+    cudaEventRecord(stop_transfer);
-    // Шаг 4: Exclusive Scan для вычисления offsets
+    cudaEventSynchronize(stop_transfer);
-    // ========================================================================
-    double step4_start = get_time_ms();
     
-    int* d_offsets = nullptr;
+    float time_transfer_ms = 0;
-    CUDA_CHECK(cudaMalloc(&d_offsets, num_periods * sizeof(int)));
+    cudaEventElapsedTime(&time_transfer_ms, start_transfer, stop_transfer);
     
-    temp_storage_bytes = 0;
+    // === ИЗМЕРЕНИЕ kernel ===
-    cub::DeviceScan::ExclusiveSum(
+    const int THREADS_PER_BLOCK = 256;
-        d_temp_storage, temp_storage_bytes,
+    int num_blocks = (num_days + THREADS_PER_BLOCK - 1) / THREADS_PER_BLOCK;
-        d_counts, d_offsets, num_periods);
     
-    CUDA_CHECK(cudaMalloc(&d_temp_storage, temp_storage_bytes));
+    cudaEventRecord(start_kernel);
     
-    cub::DeviceScan::ExclusiveSum(
+    aggregate_kernel<<<num_blocks, THREADS_PER_BLOCK>>>(
-        d_temp_storage, temp_storage_bytes,
+        d_records, num_records,
-        d_counts, d_offsets, num_periods);
+        d_day_offsets, d_day_counts, d_day_indices,
-    CUDA_CHECK(cudaGetLastError());
+        num_days, d_out_stats
+    );
     
-    cudaFree(d_temp_storage);
+    err = cudaGetLastError();
+    if (err != cudaSuccess) {
+        cudaFree(d_records);
+        cudaFree(d_day_offsets);
+        cudaFree(d_day_counts);
+        cudaFree(d_day_indices);
+        cudaFree(d_out_stats);
+        return -7;
+    }
     
-    double step4_ms = get_time_ms() - step4_start;
+    cudaEventRecord(stop_kernel);
+    cudaEventSynchronize(stop_kernel);
     
-    // ========================================================================
+    float time_kernel_ms = 0;
-    // Шаг 5: Агрегация периодов
+    cudaEventElapsedTime(&time_kernel_ms, start_kernel, stop_kernel);
-    // ========================================================================
-    double step5_start = get_time_ms();
     
-    GpuPeriodStats* d_out_stats = nullptr;
+    // === ИЗМЕРЕНИЕ memcpy D->H ===
-    CUDA_CHECK(cudaMalloc(&d_out_stats, num_periods * sizeof(GpuPeriodStats)));
+    cudaEventRecord(start_copy_back);
+    cudaMemcpy(h_out_stats, d_out_stats, num_days * sizeof(GpuDayStats), cudaMemcpyDeviceToHost);
+    cudaEventRecord(stop_copy_back);
+    cudaEventSynchronize(stop_copy_back);
     
-    // Используем простой kernel (один поток на период)
+    float time_copy_back_ms = 0;
-    // т.к. обычно тиков в периоде немного
+    cudaEventElapsedTime(&time_copy_back_ms, start_copy_back, stop_copy_back);
-    int agg_blocks = (num_periods + BLOCK_SIZE - 1) / BLOCK_SIZE;
     
-    aggregate_periods_simple_kernel<<<agg_blocks, BLOCK_SIZE>>>(
+    // === ИЗМЕРЕНИЕ cudaFree ===
-        d_open, d_high, d_low, d_close,
+    cudaEventRecord(start_free);
-        d_unique_periods, d_offsets, d_counts,
-        num_periods, d_out_stats);
-    CUDA_CHECK(cudaGetLastError());
-    CUDA_CHECK(cudaDeviceSynchronize());
     
-    double step5_ms = get_time_ms() - step5_start;
+    cudaFree(d_records);
-    
+    cudaFree(d_day_offsets);
-    // ========================================================================
+    cudaFree(d_day_counts);
-    // Шаг 6: Копирование результатов на CPU
+    cudaFree(d_day_indices);
-    // ========================================================================
-    double step6_start = get_time_ms();
-    
-    GpuPeriodStats* h_stats = new GpuPeriodStats[num_periods];
-    CUDA_CHECK(cudaMemcpy(h_stats, d_out_stats, num_periods * sizeof(GpuPeriodStats), 
-                          cudaMemcpyDeviceToHost));
-    
-    double step6_ms = get_time_ms() - step6_start;
-    
-    // ========================================================================
-    // Шаг 7: Освобождение GPU памяти
-    // ========================================================================
-    double step7_start = get_time_ms();
-    
-    cudaFree(d_timestamps);
-    cudaFree(d_open);
-    cudaFree(d_high);
-    cudaFree(d_low);
-    cudaFree(d_close);
-    cudaFree(d_period_ids);
-    cudaFree(d_unique_periods);
-    cudaFree(d_counts);
-    cudaFree(d_offsets);
-    cudaFree(d_num_runs);
     cudaFree(d_out_stats);
     
-    double step7_ms = get_time_ms() - step7_start;
+    cudaEventRecord(stop_free);
+    cudaEventSynchronize(stop_free);
     
-    // ========================================================================
+    float time_free_ms = 0;
-    // Итого
+    cudaEventElapsedTime(&time_free_ms, start_free, stop_free);
-    // ========================================================================
-    double total_ms = get_time_ms() - total_start;
     
-    // Формируем весь вывод одной строкой
+    // Общее время GPU
-    output << "  GPU aggregation (" << num_ticks << " ticks, interval=" << interval << " sec):\n";
+    float time_total_ms = time_malloc_ms + time_transfer_ms + time_kernel_ms + time_copy_back_ms + time_free_ms;
-    output << "    1. Malloc + H->D copy:  " << std::fixed << std::setprecision(3) << std::setw(7) << step1_ms << " ms\n";
-    output << "    2. Compute period_ids:  " << std::setw(7) << step2_ms << " ms\n";
-    output << "    3. RLE (CUB):           " << std::setw(7) << step3_ms << " ms (" << num_periods << " periods)\n";
-    output << "    4. Exclusive scan:      " << std::setw(7) << step4_ms << " ms\n";
-    output << "    5. Aggregation kernel:  " << std::setw(7) << step5_ms << " ms\n";
-    output << "    6. D->H copy:           " << std::setw(7) << step6_ms << " ms\n";
-    output << "    7. Free GPU memory:     " << std::setw(7) << step7_ms << " ms\n";
-    output << "    GPU TOTAL:              " << std::setw(7) << total_ms << " ms\n";
     
-    // Выводим всё одним принтом
+    // === Освобождаем события ===
-    printf("%s", output.str().c_str());
+    double cpu_event_destroy_start = get_time_ms();
+    
+    cudaEventDestroy(start_malloc);
+    cudaEventDestroy(stop_malloc);
+    cudaEventDestroy(start_transfer);
+    cudaEventDestroy(stop_transfer);
+    cudaEventDestroy(start_kernel);
+    cudaEventDestroy(stop_kernel);
+    cudaEventDestroy(start_copy_back);
+    cudaEventDestroy(stop_copy_back);
+    cudaEventDestroy(start_free);
+    cudaEventDestroy(stop_free);
+    
+    double cpu_event_destroy_ms = get_time_ms() - cpu_event_destroy_start;
+    double cpu_total_ms = get_time_ms() - cpu_total_start;
+    
+    // Выводим детальную статистику
+    printf("  GPU Timings (%d records, %d days):\n", num_records, num_days);
+    printf("    cudaMalloc:           %7.3f ms\n", time_malloc_ms);
+    printf("    memcpy H->D:          %7.3f ms\n", time_transfer_ms);
+    printf("    kernel execution:     %7.3f ms\n", time_kernel_ms);
+    printf("    memcpy D->H:          %7.3f ms\n", time_copy_back_ms);
+    printf("    cudaFree:             %7.3f ms\n", time_free_ms);
+    printf("    GPU TOTAL:            %7.3f ms\n", cpu_total_ms);
     fflush(stdout);
     
-    *h_out_stats = h_stats;
-    *out_num_periods = num_periods;
-    
     return 0;
 }
-
-// ============================================================================
-// Освобождение памяти результатов
-// ============================================================================
-
-extern "C" void gpu_free_results(GpuPeriodStats* stats) {
-    delete[] stats;
-}

src/intervals.cpp

@@ -1,301 +1,69 @@
 #include "intervals.hpp"
-#include "utils.hpp"
-#include <mpi.h>
 #include <algorithm>
 #include <cmath>
 #include <fstream>
 #include <iomanip>
 #include <sstream>
 #include <ctime>
-#include <limits>
 
-// Вспомогательная структура для накопления min/max в интервале
+std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold) {
-struct IntervalAccumulator {
+    if (days.empty()) {
-    PeriodIndex start_period;
+        return {};
-    double start_avg;
-    double open_min;
-    double open_max;
-    double close_min;
-    double close_max;
-    
-    void init(const PeriodStats& p) {
-        start_period = p.period;
-        start_avg = p.avg;
-        open_min = p.open_min;
-        open_max = p.open_max;
-        close_min = p.close_min;
-        close_max = p.close_max;
     }
     
-    void update(const PeriodStats& p) {
+    std::vector<Interval> intervals;
-        open_min = std::min(open_min, p.open_min);
-        open_max = std::max(open_max, p.open_max);
-        close_min = std::min(close_min, p.close_min);
-        close_max = std::max(close_max, p.close_max);
-    }
     
-    Interval finalize(const PeriodStats& end_period, double change) const {
-        Interval iv;
-        iv.start_period = start_period;
-        iv.end_period = end_period.period;
-        iv.start_avg = start_avg;
-        iv.end_avg = end_period.avg;
-        iv.change = change;
-        iv.open_min = std::min(open_min, end_period.open_min);
-        iv.open_max = std::max(open_max, end_period.open_max);
-        iv.close_min = std::min(close_min, end_period.close_min);
-        iv.close_max = std::max(close_max, end_period.close_max);
-        return iv;
-    }
-};
-
-// Упакованная структура PeriodStats для MPI передачи (8 doubles)
-struct PackedPeriodStats {
-    double period;   // PeriodIndex as double
-    double avg;
-    double open_min;
-    double open_max;
-    double close_min;
-    double close_max;
-    double count;    // int64_t as double
-    double valid;    // флаг валидности (1.0 = valid, 0.0 = invalid)
-    
-    void pack(const PeriodStats& ps) {
-        period = static_cast<double>(ps.period);
-        avg = ps.avg;
-        open_min = ps.open_min;
-        open_max = ps.open_max;
-        close_min = ps.close_min;
-        close_max = ps.close_max;
-        count = static_cast<double>(ps.count);
-        valid = 1.0;
-    }
-    
-    PeriodStats unpack() const {
-        PeriodStats ps;
-        ps.period = static_cast<PeriodIndex>(period);
-        ps.avg = avg;
-        ps.open_min = open_min;
-        ps.open_max = open_max;
-        ps.close_min = close_min;
-        ps.close_max = close_max;
-        ps.count = static_cast<int64_t>(count);
-        return ps;
-    }
-    
-    bool is_valid() const { return valid > 0.5; }
-    void set_invalid() { valid = 0.0; }
-};
-
-IntervalResult find_intervals_parallel(
-    const std::vector<PeriodStats>& periods,
-    int rank, int size,
-    double threshold)
-{
-    IntervalResult result;
-    result.compute_time = 0.0;
-    result.wait_time = 0.0;
-    
-    if (periods.empty()) {
-        if (rank < size - 1) {
-            PackedPeriodStats invalid;
-            invalid.set_invalid();
-            MPI_Send(&invalid, 8, MPI_DOUBLE, rank + 1, 0, MPI_COMM_WORLD);
-        }
-        return result;
-    }
-    
-    double compute_start = MPI_Wtime();
-    
-    size_t process_until = (rank == size - 1) ? periods.size() : periods.size() - 1;
-    
-    IntervalAccumulator acc;
     size_t start_idx = 0;
-    bool have_pending_interval = false;
+    double price_base = days[start_idx].avg;
     
-    if (rank > 0) {
+    for (size_t i = 1; i < days.size(); i++) {
-        double wait_start = MPI_Wtime();
+        double price_now = days[i].avg;
+        double change = std::abs(price_now - price_base) / price_base;
         
-        PackedPeriodStats received;
+        if (change >= threshold) {
-        MPI_Recv(&received, 8, MPI_DOUBLE, rank - 1, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+            Interval interval;
+            interval.start_day = days[start_idx].day;
+            interval.end_day = days[i].day;
+            interval.start_avg = price_base;
+            interval.end_avg = price_now;
+            interval.change = change;
             
-        result.wait_time = MPI_Wtime() - wait_start;
+            // Находим min/max Open и Close в интервале
-        compute_start = MPI_Wtime();
+            interval.open_min = days[start_idx].open_min;
+            interval.open_max = days[start_idx].open_max;
+            interval.close_min = days[start_idx].close_min;
+            interval.close_max = days[start_idx].close_max;
             
-        if (received.is_valid()) {
+            for (size_t j = start_idx + 1; j <= i; j++) {
-            PeriodStats prev_period = received.unpack();
+                interval.open_min = std::min(interval.open_min, days[j].open_min);
-            
+                interval.open_max = std::max(interval.open_max, days[j].open_max);
-            for (start_idx = 0; start_idx < periods.size(); start_idx++) {
+                interval.close_min = std::min(interval.close_min, days[j].close_min);
-                if (periods[start_idx].period > prev_period.period) {
+                interval.close_max = std::max(interval.close_max, days[j].close_max);
-                    break;
-                }
             }
             
-            if (start_idx < process_until) {
+            intervals.push_back(interval);
-                acc.init(prev_period);
-                have_pending_interval = true;
             
-                for (size_t i = start_idx; i < process_until; i++) {
+            // Начинаем новый интервал
-                    acc.update(periods[i]);
+            start_idx = i + 1;
-                    
+            if (start_idx >= days.size()) {
-                    double change = std::abs(periods[i].avg - acc.start_avg) / acc.start_avg;
+                break;
-                    
-                    if (change >= threshold) {
-                        result.intervals.push_back(acc.finalize(periods[i], change));
-                        have_pending_interval = false;
-                        
-                        start_idx = i + 1;
-                        if (start_idx < process_until) {
-                            acc.init(periods[start_idx]);
-                            have_pending_interval = true;
-                        }
-                    }
-                }
             }
-        } else {
+            price_base = days[start_idx].avg;
-            if (process_until > 0) {
-                acc.init(periods[0]);
-                have_pending_interval = true;
-                start_idx = 0;
-            }
-        }
-    } else {
-        if (process_until > 0) {
-            acc.init(periods[0]);
-            have_pending_interval = true;
-            start_idx = 0;
         }
     }
     
-    if (rank == 0 && have_pending_interval) {
+    return intervals;
-        for (size_t i = 1; i < process_until; i++) {
-            acc.update(periods[i]);
-            
-            double change = std::abs(periods[i].avg - acc.start_avg) / acc.start_avg;
-            
-            if (change >= threshold) {
-                result.intervals.push_back(acc.finalize(periods[i], change));
-                have_pending_interval = false;
-                
-                start_idx = i + 1;
-                if (start_idx < process_until) {
-                    acc.init(periods[start_idx]);
-                    have_pending_interval = true;
-                }
-            }
-        }
-    }
-    
-    if (rank == size - 1 && have_pending_interval && !periods.empty()) {
-        const auto& last_period = periods.back();
-        double change = std::abs(last_period.avg - acc.start_avg) / acc.start_avg;
-        result.intervals.push_back(acc.finalize(last_period, change));
-    }
-    
-    result.compute_time = MPI_Wtime() - compute_start;
-    
-    if (rank < size - 1) {
-        PackedPeriodStats to_send;
-        
-        if (have_pending_interval) {
-            PeriodStats start_period;
-            start_period.period = acc.start_period;
-            start_period.avg = acc.start_avg;
-            start_period.open_min = acc.open_min;
-            start_period.open_max = acc.open_max;
-            start_period.close_min = acc.close_min;
-            start_period.close_max = acc.close_max;
-            start_period.count = 0;
-            to_send.pack(start_period);
-        } else if (periods.size() >= 2) {
-            to_send.pack(periods[periods.size() - 2]);
-        } else {
-            to_send.set_invalid();
-        }
-        
-        MPI_Send(&to_send, 8, MPI_DOUBLE, rank + 1, 0, MPI_COMM_WORLD);
-    }
-    
-    return result;
 }
 
-double collect_intervals(
+std::string day_index_to_date(DayIndex day) {
-    std::vector<Interval>& local_intervals,
+    time_t ts = static_cast<time_t>(day) * 86400;
-    int rank, int size)
-{
-    double wait_time = 0.0;
-    
-    if (rank == 0) {
-        for (int r = 1; r < size; r++) {
-            double wait_start = MPI_Wtime();
-            
-            int count;
-            MPI_Recv(&count, 1, MPI_INT, r, 1, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-            
-            if (count > 0) {
-                std::vector<double> buffer(count * 9);
-                MPI_Recv(buffer.data(), count * 9, MPI_DOUBLE, r, 2, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-                
-                for (int i = 0; i < count; i++) {
-                    Interval iv;
-                    iv.start_period = static_cast<PeriodIndex>(buffer[i * 9 + 0]);
-                    iv.end_period = static_cast<PeriodIndex>(buffer[i * 9 + 1]);
-                    iv.open_min = buffer[i * 9 + 2];
-                    iv.open_max = buffer[i * 9 + 3];
-                    iv.close_min = buffer[i * 9 + 4];
-                    iv.close_max = buffer[i * 9 + 5];
-                    iv.start_avg = buffer[i * 9 + 6];
-                    iv.end_avg = buffer[i * 9 + 7];
-                    iv.change = buffer[i * 9 + 8];
-                    local_intervals.push_back(iv);
-                }
-            }
-            
-            wait_time += MPI_Wtime() - wait_start;
-        }
-        
-        std::sort(local_intervals.begin(), local_intervals.end(),
-            [](const Interval& a, const Interval& b) {
-                return a.start_period < b.start_period;
-            });
-    } else {
-        int count = static_cast<int>(local_intervals.size());
-        MPI_Send(&count, 1, MPI_INT, 0, 1, MPI_COMM_WORLD);
-        
-        if (count > 0) {
-            std::vector<double> buffer(count * 9);
-            for (int i = 0; i < count; i++) {
-                const auto& iv = local_intervals[i];
-                buffer[i * 9 + 0] = static_cast<double>(iv.start_period);
-                buffer[i * 9 + 1] = static_cast<double>(iv.end_period);
-                buffer[i * 9 + 2] = iv.open_min;
-                buffer[i * 9 + 3] = iv.open_max;
-                buffer[i * 9 + 4] = iv.close_min;
-                buffer[i * 9 + 5] = iv.close_max;
-                buffer[i * 9 + 6] = iv.start_avg;
-                buffer[i * 9 + 7] = iv.end_avg;
-                buffer[i * 9 + 8] = iv.change;
-            }
-            MPI_Send(buffer.data(), count * 9, MPI_DOUBLE, 0, 2, MPI_COMM_WORLD);
-        }
-    }
-    
-    return wait_time;
-}
-
-std::string period_index_to_datetime(PeriodIndex period) {
-    int64_t interval = get_aggregation_interval();
-    time_t ts = static_cast<time_t>(period) * interval;
     struct tm* tm_info = gmtime(&ts);
     
     std::ostringstream oss;
     oss << std::setfill('0') 
         << (tm_info->tm_year + 1900) << "-"
         << std::setw(2) << (tm_info->tm_mon + 1) << "-"
-        << std::setw(2) << tm_info->tm_mday << " "
+        << std::setw(2) << tm_info->tm_mday;
-        << std::setw(2) << tm_info->tm_hour << ":"
-        << std::setw(2) << tm_info->tm_min << ":"
-        << std::setw(2) << tm_info->tm_sec;
     
     return oss.str();
 }
@@ -304,11 +72,11 @@ void write_intervals(const std::string& filename, const std::vector<Interval>& i
     std::ofstream out(filename);
     
     out << std::fixed << std::setprecision(2);
-    out << "start_datetime,end_datetime,open_min,open_max,close_min,close_max,start_avg,end_avg,change\n";
+    out << "start_date,end_date,open_min,open_max,close_min,close_max,start_avg,end_avg,change\n";
     
     for (const auto& iv : intervals) {
-        out << period_index_to_datetime(iv.start_period) << ","
+        out << day_index_to_date(iv.start_day) << ","
-            << period_index_to_datetime(iv.end_period) << ","
+            << day_index_to_date(iv.end_day) << ","
             << iv.open_min << ","
             << iv.open_max << ","
             << iv.close_min << ","

src/intervals.hpp

@@ -1,44 +1,27 @@
 #pragma once
 
-#include "period_stats.hpp"
+#include "day_stats.hpp"
 #include <vector>
 #include <string>
 
 // Интервал с изменением >= threshold
 struct Interval {
-    PeriodIndex start_period;
+    DayIndex start_day;
-    PeriodIndex end_period;
+    DayIndex end_day;
-    double open_min;
+    double open_min;    // минимальный Open в интервале
-    double open_max;
+    double open_max;    // максимальный Open в интервале
-    double close_min;
+    double close_min;   // минимальный Close в интервале
-    double close_max;
+    double close_max;   // максимальный Close в интервале
     double start_avg;
     double end_avg;
     double change;
 };
 
-// Результат параллельного построения интервалов
+// Вычисление интервалов с изменением >= threshold (по умолчанию 10%)
-struct IntervalResult {
+std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold = 0.10);
-    std::vector<Interval> intervals;
-    double compute_time;  // время вычислений
-    double wait_time;     // время ожидания данных от предыдущего ранка
-};
-
-// Параллельное построение интервалов с использованием MPI
-IntervalResult find_intervals_parallel(
-    const std::vector<PeriodStats>& periods,
-    int rank, int size,
-    double threshold = 0.10
-);
-
-// Сбор интервалов со всех ранков на ранк 0
-double collect_intervals(
-    std::vector<Interval>& local_intervals,
-    int rank, int size
-);
 
 // Вывод интервалов в файл
 void write_intervals(const std::string& filename, const std::vector<Interval>& intervals);
 
-// Преобразование PeriodIndex в строку даты/времени
+// Преобразование DayIndex в строку даты (YYYY-MM-DD)
-std::string period_index_to_datetime(PeriodIndex period);
+std::string day_index_to_date(DayIndex day);

src/main.cpp

@@ -5,31 +5,16 @@
 
 #include "csv_loader.hpp"
 #include "record.hpp"
-#include "period_stats.hpp"
+#include "day_stats.hpp"
 #include "aggregation.hpp"
-#include "intervals.hpp"
-#include "utils.hpp"
-#include "gpu_loader.hpp"
 
 int main(int argc, char** argv) {
     MPI_Init(&argc, &argv);
-    double total_start = MPI_Wtime();
 
     int rank, size;
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
 
-    // Проверяем доступность GPU
-    bool use_cuda = get_use_cuda();
-    bool have_gpu = gpu_is_available();
-    bool use_gpu = use_cuda && have_gpu;
-    
-    std::cout << "Rank " << rank 
-              << ": USE_CUDA=" << use_cuda 
-              << ", GPU available=" << have_gpu 
-              << ", using " << (use_gpu ? "GPU" : "CPU")
-              << std::endl;
-
     // Параллельное чтение данных
     double read_start = MPI_Wtime();
     std::vector<Record> records = load_csv_parallel(rank, size);
@@ -40,76 +25,16 @@ int main(int argc, char** argv) {
               << " in " << std::fixed << std::setprecision(3) << read_time << " sec"
               << std::endl;
 
-    // Агрегация по периодам
+    // Агрегация по дням
     double agg_start = MPI_Wtime();
-    std::vector<PeriodStats> periods;
+    std::vector<DayStats> days = aggregate_days(records);
-    
-    if (use_gpu) {
-        int64_t interval = get_aggregation_interval();
-        if (!aggregate_periods_gpu(records, interval, periods)) {
-            std::cerr << "Rank " << rank << ": GPU aggregation failed, falling back to CPU" << std::endl;
-            periods = aggregate_periods(records);
-        }
-    } else {
-        periods = aggregate_periods(records);
-    }
-    
     double agg_time = MPI_Wtime() - agg_start;
 
     std::cout << "Rank " << rank 
-              << ": aggregated " << periods.size() << " periods"
+              << ": aggregated " << days.size() << " days"
-              << " [" << (periods.empty() ? 0 : periods.front().period) 
-              << ".." << (periods.empty() ? 0 : periods.back().period) << "]"
               << " in " << std::fixed << std::setprecision(3) << agg_time << " sec"
               << std::endl;
 
-    // Удаляем крайние периоды (могут быть неполными из-за параллельного чтения)
-    trim_edge_periods(periods, rank, size);
-
-    std::cout << "Rank " << rank 
-              << ": after trim " << periods.size() << " periods"
-              << " [" << (periods.empty() ? 0 : periods.front().period) 
-              << ".." << (periods.empty() ? 0 : periods.back().period) << "]"
-              << std::endl;
-
-    // Параллельное построение интервалов
-    IntervalResult iv_result = find_intervals_parallel(periods, rank, size);
-
-    std::cout << "Rank " << rank 
-              << ": found " << iv_result.intervals.size() << " intervals"
-              << ", compute " << std::fixed << std::setprecision(6) << iv_result.compute_time << " sec"
-              << ", wait " << iv_result.wait_time << " sec"
-              << std::endl;
-
-    // Сбор интервалов на ранке 0
-    double collect_wait = collect_intervals(iv_result.intervals, rank, size);
-
-    if (rank == 0) {
-        std::cout << "Rank 0: collected " << iv_result.intervals.size() << " total intervals"
-                  << ", wait " << std::fixed << std::setprecision(3) << collect_wait << " sec"
-                  << std::endl;
-    }
-
-    // Запись результатов в файл (только ранк 0)
-    if (rank == 0) {
-        double write_start = MPI_Wtime();
-        write_intervals("result.csv", iv_result.intervals);
-        double write_time = MPI_Wtime() - write_start;
-
-        std::cout << "Rank 0: wrote result.csv"
-                  << " in " << std::fixed << std::setprecision(3) << write_time << " sec"
-                  << std::endl;
-    }
-
-    // Вывод общего времени выполнения
-    MPI_Barrier(MPI_COMM_WORLD);
-    double total_time = MPI_Wtime() - total_start;
-    if (rank == 0) {
-        std::cout << "Total execution time: "
-                  << std::fixed << std::setprecision(3)
-                  << total_time << " sec" << std::endl;
-    }
-
     MPI_Finalize();
     return 0;
 }

src/period_stats.hpp

@@ -1,15 +0,0 @@
-#pragma once
-#include <cstdint>
-
-using PeriodIndex = int64_t;
-
-// Агрегированные данные за один период
-struct PeriodStats {
-    PeriodIndex period;   // индекс периода (timestamp / AGGREGATION_INTERVAL)
-    double avg;           // среднее значение (Low + High) / 2 по всем записям
-    double open_min;      // минимальный Open за период
-    double open_max;      // максимальный Open за период
-    double close_min;     // минимальный Close за период
-    double close_max;     // максимальный Close за период
-    int64_t count;        // количество записей, по которым агрегировали
-};

src/utils.cpp

@@ -4,6 +4,29 @@
 #include <stdexcept>
 #include <numeric>
 
+std::map<DayIndex, std::vector<Record>> group_by_day(const std::vector<Record>& recs) {
+    std::map<DayIndex, std::vector<Record>> days;
+
+    for (const auto& r : recs) {
+        DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
+        days[day].push_back(r);
+    }
+
+    return days;
+}
+
+std::vector<std::vector<DayIndex>> split_days(const std::map<DayIndex, std::vector<Record>>& days, int parts) {
+    std::vector<std::vector<DayIndex>> out(parts);
+
+    int i = 0;
+    for (auto& kv : days) {
+        out[i % parts].push_back(kv.first);
+        i++;
+    }
+
+    return out;
+}
+
 int get_num_cpu_threads() {
     const char* env_threads = std::getenv("NUM_CPU_THREADS");
     int num_cpu_threads = 1;
@@ -40,14 +63,6 @@ int64_t get_read_overlap_bytes() {
     return std::stoll(get_env("READ_OVERLAP_BYTES"));
 }
 
-int64_t get_aggregation_interval() {
-    return std::stoll(get_env("AGGREGATION_INTERVAL"));
-}
-
-bool get_use_cuda() {
-    return std::stoi(get_env("USE_CUDA")) != 0;
-}
-
 int64_t get_file_size(const std::string& path) {
     std::ifstream file(path, std::ios::binary | std::ios::ate);
     if (!file.is_open()) {
@@ -58,6 +73,7 @@ int64_t get_file_size(const std::string& path) {
 
 ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
                                const std::vector<int>& shares, int64_t overlap_bytes) {
+    // Если shares пустой или не соответствует size, используем равные доли
     std::vector<int> effective_shares;
     if (shares.size() == static_cast<size_t>(size)) {
         effective_shares = shares;
@@ -66,6 +82,8 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
     }
     
     int total_shares = std::accumulate(effective_shares.begin(), effective_shares.end(), 0);
+    
+    // Вычисляем базовые границы для каждого ранка
     int64_t bytes_per_share = file_size / total_shares;
     
     int64_t base_start = 0;
@@ -75,31 +93,22 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
     
     int64_t base_end = base_start + bytes_per_share * effective_shares[rank];
     
+    // Применяем overlap
     ByteRange range;
     
     if (rank == 0) {
+        // Первый ранк: начинаем с 0, добавляем overlap в конце
         range.start = 0;
         range.end = std::min(base_end + overlap_bytes, file_size);
     } else if (rank == size - 1) {
+        // Последний ранк: вычитаем overlap в начале, читаем до конца файла
         range.start = std::max(base_start - overlap_bytes, static_cast<int64_t>(0));
         range.end = file_size;
     } else {
+        // Промежуточные ранки: overlap с обеих сторон
         range.start = std::max(base_start - overlap_bytes, static_cast<int64_t>(0));
         range.end = std::min(base_end + overlap_bytes, file_size);
     }
     
     return range;
 }
-
-void trim_edge_periods(std::vector<PeriodStats>& periods, int rank, int size) {
-    if (periods.empty()) return;
-    
-    if (rank == 0) {
-        periods.pop_back();
-    } else if (rank == size - 1) {
-        periods.erase(periods.begin());
-    } else {
-        periods.pop_back();
-        periods.erase(periods.begin());
-    }
-}

src/utils.hpp

@@ -1,20 +1,22 @@
 #pragma once
 
 #include "record.hpp"
-#include "period_stats.hpp"
+#include "day_stats.hpp"
 #include <map>
 #include <vector>
 #include <string>
 #include <cstdlib>
 #include <cstdint>
 
+// Группировка записей по дням
+std::map<DayIndex, std::vector<Record>> group_by_day(const std::vector<Record>& recs);
+std::vector<std::vector<DayIndex>> split_days(const std::map<DayIndex, std::vector<Record>>& days, int parts);
+
 // Чтение переменных окружения
 int get_num_cpu_threads();
 std::string get_data_path();
 std::vector<int> get_data_read_shares();
 int64_t get_read_overlap_bytes();
-int64_t get_aggregation_interval();
-bool get_use_cuda();
 
 // Структура для хранения диапазона байт для чтения
 struct ByteRange {
@@ -28,6 +30,3 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
 
 // Получение размера файла
 int64_t get_file_size(const std::string& path);
-
-// Удаляет крайние периоды, которые могут быть неполными из-за параллельного чтения
-void trim_edge_periods(std::vector<PeriodStats>& periods, int rank, int size);

upsample.py

@@ -1,136 +0,0 @@
-#!/usr/bin/env python3
-import argparse
-import sys
-import time
-
-
-def parse_row(line: str):
-    # Timestamp,Open,High,Low,Close,Volume
-    ts, o, h, l, c, v = line.split(',')
-    return int(float(ts)), float(o), float(h), float(l), float(c), float(v)
-
-
-def fmt_row(ts, o, h, l, c, v):
-    return f"{ts},{o:.2f},{h:.2f},{l:.2f},{c:.2f},{v:.8f}\n"
-
-
-def count_lines_fast(path: str) -> int:
-    with open(path, "rb") as f:
-        return sum(1 for _ in f) - 1  # минус header
-
-
-def main(inp, out, step, flush_every):
-    # считаем количество строк для прогресса
-    total_lines = count_lines_fast(inp)
-    print(f"Total input rows: {total_lines:,}", file=sys.stderr)
-
-    start_time = time.time()
-    processed = 0
-    last_report = start_time
-
-    with open(inp, "r", buffering=8 * 1024 * 1024) as fin, \
-         open(out, "w", buffering=8 * 1024 * 1024) as fout:
-
-        fin.readline()  # пропускаем header
-        fout.write("Timestamp,Open,High,Low,Close,Volume\n")
-
-        first = fin.readline()
-        if not first:
-            return
-
-        prev = parse_row(first.strip())
-
-        out_buf = []
-        out_rows = 0
-
-        for line in fin:
-            line = line.strip()
-            if not line:
-                continue
-
-            cur = parse_row(line)
-
-            t1, o1, h1, l1, c1, v1 = prev
-            t2, o2, h2, l2, c2, v2 = cur
-
-            dt = t2 - t1
-            steps = dt // step
-
-            if steps > 0:
-                do = o2 - o1
-                dh = h2 - h1
-                dl = l2 - l1
-                dc = c2 - c1
-                dv = v2 - v1
-
-                inv = 1.0 / steps
-                for i in range(steps):
-                    a = i * inv
-                    out_buf.append(fmt_row(
-                        t1 + i * step,
-                        o1 + do * a,
-                        h1 + dh * a,
-                        l1 + dl * a,
-                        c1 + dc * a,
-                        v1 + dv * a
-                    ))
-
-                out_rows += steps
-
-            prev = cur
-            processed += 1
-
-            # прогресс
-            if processed % 100_000 == 0:
-                now = time.time()
-                if now - last_report >= 0.5:
-                    pct = processed * 100.0 / total_lines
-                    elapsed = now - start_time
-                    speed = processed / elapsed if elapsed > 0 else 0
-                    eta = (total_lines - processed) / speed if speed > 0 else 0
-
-                    print(
-                        f"\rprocessed: {processed:,} / {total_lines:,} "
-                        f"({pct:5.1f}%) | "
-                        f"out ~ {out_rows:,} | "
-                        f"{speed:,.0f} rows/s | "
-                        f"ETA {eta/60:5.1f} min",
-                        end="",
-                        file=sys.stderr,
-                        flush=True,
-                    )
-                    last_report = now
-
-            # сброс буфера
-            if out_rows >= flush_every:
-                fout.write("".join(out_buf))
-                out_buf.clear()
-                out_rows = 0
-
-        # остатки
-        if out_buf:
-            fout.write("".join(out_buf))
-
-        # последнюю строку пишем как есть
-        t, o, h, l, c, v = prev
-        fout.write(fmt_row(t, o, h, l, c, v))
-
-    total_time = time.time() - start_time
-    print(
-        f"\nDone in {total_time/60:.1f} min",
-        file=sys.stderr
-    )
-
-
-if __name__ == "__main__":
-    ap = argparse.ArgumentParser()
-    ap.add_argument("-i", "--input", required=True)
-    ap.add_argument("-o", "--output", required=True)
-    ap.add_argument("-s", "--step", type=int, default=10)
-    ap.add_argument("--flush-every", type=int, default=200_000)
-    args = ap.parse_args()
-
-    if args.step <= 0:
-        raise SystemExit("step must be > 0")
-
-    main(args.input, args.output, args.step, args.flush_every)