На CPU вычисления

.gitignore

@@ -1,3 +1,4 @@
 data
 build
 out.txt
+result.csv

src/aggregation.cpp

@@ -0,0 +1,87 @@
+#include "aggregation.hpp"
+#include <algorithm>
+#include <limits>
+#include <cmath>
+
+std::vector<DayStats> aggregate_days(const std::vector<Record>& records) {
+    // Группируем записи по дням
+    std::map<DayIndex, std::vector<const Record*>> day_records;
+    
+    for (const auto& r : records) {
+        DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
+        day_records[day].push_back(&r);
+    }
+    
+    std::vector<DayStats> result;
+    result.reserve(day_records.size());
+    
+    for (auto& [day, recs] : day_records) {
+        // Сортируем по timestamp для определения first/last
+        std::sort(recs.begin(), recs.end(), 
+            [](const Record* a, const Record* b) {
+                return a->timestamp < b->timestamp;
+            });
+        
+        DayStats stats;
+        stats.day = day;
+        stats.low = std::numeric_limits<double>::max();
+        stats.high = std::numeric_limits<double>::lowest();
+        stats.open = recs.front()->open;
+        stats.close = recs.back()->close;
+        stats.first_ts = recs.front()->timestamp;
+        stats.last_ts = recs.back()->timestamp;
+        
+        for (const auto* r : recs) {
+            stats.low = std::min(stats.low, r->low);
+            stats.high = std::max(stats.high, r->high);
+        }
+        
+        stats.avg = (stats.low + stats.high) / 2.0;
+        
+        result.push_back(stats);
+    }
+    
+    return result;
+}
+
+std::vector<DayStats> merge_day_stats(const std::vector<DayStats>& all_stats) {
+    // Объединяем статистику по одинаковым дням (если такие есть)
+    std::map<DayIndex, DayStats> merged;
+    
+    for (const auto& s : all_stats) {
+        auto it = merged.find(s.day);
+        if (it == merged.end()) {
+            merged[s.day] = s;
+        } else {
+            // Объединяем данные за один день
+            auto& m = it->second;
+            m.low = std::min(m.low, s.low);
+            m.high = std::max(m.high, s.high);
+            
+            // open берём от записи с меньшим timestamp
+            if (s.first_ts < m.first_ts) {
+                m.open = s.open;
+                m.first_ts = s.first_ts;
+            }
+            
+            // close берём от записи с большим timestamp
+            if (s.last_ts > m.last_ts) {
+                m.close = s.close;
+                m.last_ts = s.last_ts;
+            }
+            
+            m.avg = (m.low + m.high) / 2.0;
+        }
+    }
+    
+    // Преобразуем в отсортированный вектор
+    std::vector<DayStats> result;
+    result.reserve(merged.size());
+    
+    for (auto& [day, stats] : merged) {
+        result.push_back(stats);
+    }
+    
+    return result;
+}
+

src/aggregation.hpp

@@ -0,0 +1,14 @@
+#pragma once
+
+#include "record.hpp"
+#include "day_stats.hpp"
+#include <vector>
+#include <map>
+
+// Агрегация записей по дням на одном узле
+std::vector<DayStats> aggregate_days(const std::vector<Record>& records);
+
+// Объединение агрегированных данных с разных узлов
+// (на случай если один день попал на разные узлы - но в нашей схеме это не должно случиться)
+std::vector<DayStats> merge_day_stats(const std::vector<DayStats>& all_stats);
+

src/day_stats.hpp

@@ -0,0 +1,28 @@
+#pragma once
+#include <cstdint>
+
+using DayIndex = long long;
+
+// Агрегированные данные за один день
+struct DayStats {
+    DayIndex day;       // индекс дня (timestamp / 86400)
+    double low;         // минимальный Low за день
+    double high;        // максимальный High за день
+    double open;        // первый Open за день
+    double close;       // последний Close за день
+    double avg;         // среднее = (low + high) / 2
+    double first_ts;    // timestamp первой записи (для определения порядка open)
+    double last_ts;     // timestamp последней записи (для определения close)
+};
+
+// Интервал с изменением >= 10%
+struct Interval {
+    DayIndex start_day;
+    DayIndex end_day;
+    double min_open;
+    double max_close;
+    double start_avg;
+    double end_avg;
+    double change;
+};
+

src/intervals.cpp

@@ -0,0 +1,83 @@
+#include "intervals.hpp"
+#include <algorithm>
+#include <cmath>
+#include <fstream>
+#include <iomanip>
+#include <sstream>
+#include <ctime>
+
+std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold) {
+    if (days.empty()) {
+        return {};
+    }
+    
+    std::vector<Interval> intervals;
+    
+    size_t start_idx = 0;
+    double price_base = days[start_idx].avg;
+    
+    for (size_t i = 1; i < days.size(); i++) {
+        double price_now = days[i].avg;
+        double change = std::abs(price_now - price_base) / price_base;
+        
+        if (change >= threshold) {
+            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;
+            
+            // Находим min(Open) и max(Close) в интервале
+            interval.min_open = days[start_idx].open;
+            interval.max_close = days[start_idx].close;
+            
+            for (size_t j = start_idx; j <= i; j++) {
+                interval.min_open = std::min(interval.min_open, days[j].open);
+                interval.max_close = std::max(interval.max_close, days[j].close);
+            }
+            
+            intervals.push_back(interval);
+            
+            // Начинаем новый интервал
+            start_idx = i + 1;
+            if (start_idx >= days.size()) {
+                break;
+            }
+            price_base = days[start_idx].avg;
+        }
+    }
+    
+    return intervals;
+}
+
+std::string day_index_to_date(DayIndex day) {
+    time_t ts = static_cast<time_t>(day) * 86400;
+    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;
+    
+    return oss.str();
+}
+
+void write_intervals(const std::string& filename, const std::vector<Interval>& intervals) {
+    std::ofstream out(filename);
+    
+    out << std::fixed << std::setprecision(2);
+    out << "start_date,end_date,min_open,max_close,start_avg,end_avg,change\n";
+    
+    for (const auto& iv : intervals) {
+        out << day_index_to_date(iv.start_day) << ","
+            << day_index_to_date(iv.end_day) << ","
+            << iv.min_open << ","
+            << iv.max_close << ","
+            << iv.start_avg << ","
+            << iv.end_avg << ","
+            << std::setprecision(6) << iv.change << "\n";
+    }
+}
+

src/intervals.hpp

@@ -0,0 +1,15 @@
+#pragma once
+
+#include "day_stats.hpp"
+#include <vector>
+#include <string>
+
+// Вычисление интервалов с изменением >= threshold (по умолчанию 10%)
+std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold = 0.10);
+
+// Вывод интервалов в файл
+void write_intervals(const std::string& filename, const std::vector<Interval>& intervals);
+
+// Преобразование DayIndex в строку даты (YYYY-MM-DD)
+std::string day_index_to_date(DayIndex day);
+

src/main.cpp

@@ -6,12 +6,15 @@
 #include "csv_loader.hpp"
 #include "utils.hpp"
 #include "record.hpp"
+#include "day_stats.hpp"
+#include "aggregation.hpp"
+#include "intervals.hpp"
 #include "gpu_loader.hpp"
 
 // Функция: отобрать записи для конкретного ранга
 std::vector<Record> select_records_for_rank(
-    const std::map<long long, std::vector<Record>>& days,
-    const std::vector<long long>& day_list)
+    const std::map<DayIndex, std::vector<Record>>& days,
+    const std::vector<DayIndex>& day_list)
 {
     std::vector<Record> out;
     for (auto d : day_list) {
@@ -52,7 +55,7 @@ int main(int argc, char** argv) {
                 continue;
             }
 
-            int count = vec.size();
+            int count = static_cast<int>(vec.size());
             MPI_Send(&count, 1, MPI_INT, r, 0, MPI_COMM_WORLD);
             MPI_Send(vec.data(), count * sizeof(Record), MPI_BYTE, r, 1, MPI_COMM_WORLD);
         }
@@ -72,14 +75,72 @@ int main(int argc, char** argv) {
     std::cout << "Rank " << rank << " received "
               << local_records.size() << " records" << std::endl;
 
+    // ====== АГРЕГАЦИЯ НА КАЖДОМ УЗЛЕ ======
+    auto local_stats = aggregate_days(local_records);
+    std::cout << "Rank " << rank << " aggregated " 
+              << local_stats.size() << " days" << std::endl;
 
+    // ====== СБОР АГРЕГИРОВАННЫХ ДАННЫХ НА RANK 0 ======
+    std::vector<DayStats> all_stats;
+
+    if (rank == 0) {
+        // Добавляем свои данные
+        all_stats.insert(all_stats.end(), local_stats.begin(), local_stats.end());
+
+        // Получаем данные от других узлов
+        for (int r = 1; r < size; r++) {
+            int count = 0;
+            MPI_Recv(&count, 1, MPI_INT, r, 2, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+
+            std::vector<DayStats> remote_stats(count);
+            MPI_Recv(remote_stats.data(), count * sizeof(DayStats),
+                     MPI_BYTE, r, 3, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+
+            all_stats.insert(all_stats.end(), remote_stats.begin(), remote_stats.end());
+        }
+    } else {
+        // Отправляем свои агрегированные данные на rank 0
+        int count = static_cast<int>(local_stats.size());
+        MPI_Send(&count, 1, MPI_INT, 0, 2, MPI_COMM_WORLD);
+        MPI_Send(local_stats.data(), count * sizeof(DayStats), MPI_BYTE, 0, 3, MPI_COMM_WORLD);
+    }
+
+    // ====== ВЫЧИСЛЕНИЕ ИНТЕРВАЛОВ НА RANK 0 ======
+    if (rank == 0) {
+        std::cout << "Rank 0: merging " << all_stats.size() << " day stats..." << std::endl;
+
+        // Объединяем и сортируем
+        auto merged_stats = merge_day_stats(all_stats);
+        std::cout << "Rank 0: total " << merged_stats.size() << " unique days" << std::endl;
+
+        // Вычисляем интервалы
+        auto intervals = find_intervals(merged_stats, 0.10);
+        std::cout << "Found " << intervals.size() << " intervals with >=10% change" << std::endl;
+
+        // Записываем результат
+        write_intervals("../result.csv", intervals);
+        std::cout << "Results written to result.csv" << std::endl;
+
+        // Выводим первые несколько интервалов
+        std::cout << "\nFirst 5 intervals:\n";
+        std::cout << "start_date,end_date,min_open,max_close,change\n";
+        for (size_t i = 0; i < std::min(intervals.size(), size_t(5)); i++) {
+            const auto& iv = intervals[i];
+            std::cout << day_index_to_date(iv.start_day) << ","
+                      << day_index_to_date(iv.end_day) << ","
+                      << iv.min_open << ","
+                      << iv.max_close << ","
+                      << iv.change << "\n";
+        }
+    }
+
+    // Проверка GPU (оставляем как есть)
     auto gpu_is_available = load_gpu_is_available();
     int have_gpu = 0;
     if (gpu_is_available) {
         std::cout << "Rank " << rank << " dll loaded" << std::endl;
         have_gpu = gpu_is_available();
     }
-
     std::cout << "Rank " << rank << ": gpu_available=" << have_gpu << "\n";
 
     MPI_Finalize();

src/utils.hpp

@@ -1,11 +1,9 @@
 #pragma once
 
 #include "record.hpp"
+#include "day_stats.hpp"
 #include <map>
 #include <vector>
 
-using DayIndex = long long;
-
 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);
-