shanks-university

Поток данных (Dataflow)

Обзор

Документ описывает полный жизненный цикл данных в фреймворке - от загрузки конфигурации до экспорта результатов и визуализации.

Пошаговый поток данных

Шаг 1. Инициализация (CLI)

Файл: cli.py

python3 run --config backend/runner/config/options.json

Шаг 2. Загрузка конфигурации

Файл: config/model.py

cfg = TrialConfig.load(Path("config/options.json"))

Загружаемые поля:

Поле	Описание	По умолчанию
`verbose`	Уровень логирования	0
`series_json`	Путь к JSON с рядами	`config/example.json`
`accel_json`	Путь к JSON с методами ускорения	`config/example.json`
`noise_json`	Путь к JSON с шумами	`config/example.json`
`filters_json`	Путь к JSON с фильтрами	`config/example.json`
`output_dir`	Директория вывода	`output`
`results_filename`	Имя файла результатов	`results`
`trial_process_count`	Число параллельных процессов	1
`trial_task_timeout`	Таймаут задачи (сек)	10
`trial_memory_efficient`	Режим экономии памяти	true
`precisions`	Типы точности	`[F64]`
`output_formats`	Форматы вывода	`[JSON, CSV]`

Дополнительно загружается:

noises — список конфигураций шума из noise_json
filters — список конфигураций фильтров из filters_json

Шаг 3. Dependency Injection

Файл: dependency.py

executor = get_trial_executor_from_config(cfg)

Создаётся:

TrialRunner — ParallelTrialRunner или SequentialTrialRunner в зависимости от trial_process_count
SeriesParamSource’ы — из PARAM_SERIES_SOURCE_BUILDERS registry:
- JSONSeriesParamSource
- CSVSeriesParamSource
AccelParamSource’ы — из PARAM_ACCELS_SOURCE_BUILDERS registry:
- JSONAccelParamSource — если accel_json существует
Serializer — TrialResultSerializer
Exporters — из EXPORT_BUILDERS registry для каждого формата в output_formats

Результат: Fully configured TrialExecutor

Шаг 4. Загрузка параметров

Файл: services/trial_executor.py:load_parameters()

series_params, accel_params = executor.load_parameters(PrecisionType.F64)

Процесс:

Для каждого SeriesParamSource вызывается load(precision)
Для каждого AccelParamSource вызывается load(precision)
Источники также загружают noise_configs

Из JSON файла загружаются:

Series параметры

{
  "series": [
    {
      "name": "CosSeries",
      "args": {"x": [1, 2]}
    }
  ]
}

Accel параметры

{
  "methods": [
    {
      "name": "LevinAlgorithm",
      "n": {"start": 10, "stop": 100, "step": 10},
      "m": [8],
      "args": {
        "remainder": ["v_type"],
        "useRecurrentFormula": true
      },
      "events": [...]
    }
  ]
}

Функция autowrap:

Единичное значение -> одно значение
Список -> каждое значение отдельно
Объект {start, stop, step} -> генерирует диапазон

Результат:

series_params: list[BaseSeriesParam]
accel_params: list[BaseAccelParam]

Шаг 5. Генерация комбинаций

Файл: domain/complex_trial.py

trial = ComplexTrial(series_params, accel_params)
combinations = trial.combinations()

Декартово произведение:

series_params * accel_params = [
    (series_1, accel_1),
    (series_1, accel_2),
    ...
    (series_N, accel_M)
]

Каждая комбинация будет выполнена со всеми:

noise_configs (если есть)
filter_configs (если есть)
precisions (итерация внешнего цикла)

Результат: list[tuple[BaseSeriesParam, BaseAccelParam]]

Шаг 6. Выполнение trial

Файл: domain/use_cases/run_trial.py:execute_trial()

results = execute_trial(
    (series, accel),
    noise_config,
    filter_configs
)

Подпроцессы:

6.1 Получение ряда

series_result, series_lim = series.obtain_by_argument(series_argument, size_floor)

Создаётся инстанс C++ ряда (например CosSeriesF64)
Генерируются значения ряда до size_floor
Вычисляется теоретический предел series_lim

6.2 Применение шума (опционально)

if noise_config:
    func_name = f"applyNoise{precision.value}"
    series_result = getattr(ps, func_name)(
        series_result,
        noise_method_enum,
        noise_type_enum,
        seed, param1, param2
    )

6.3 Создание инстанса алгоритма

accel_instance = accel.create_instance(additional_args)

Создаётся C++ инстанс алгоритма (например LevinAlgorithmF64)

6.4 Итерация по n и m

for m_value in m_values:
    ctx = accel.create_event_context()
    for n_value in n_values:
        partial_sum = series_result.Sn[n_value - 1]
        accel_value = accel_instance(n_value, m_value, series_result)

        computed.append(ComputedTrialResult(...))
        events = accel.process_events(computed, ctx)

Вычисляемые метрики:

series_value - член ряда a_n
partial_sum - частичная сумма S_n
partial_sum_deviation - S_n - lim
accel_value - значение ускорения A_n
accel_value_deviation - A_n - lim
events - список обнаруженных событий

6.5 Детекция событий

Файл: domain/event.py

Типы событий:

slow_accel - A_n - lim > S_n - lim
monotone - монотонность
divergent - расходимость
sign_changed - смена знака
second_diff - рост второй разности

Обработка событий:

log_action_capacity - максимальное число раз, которое можно логировавть событий.
stop_action_limit - остановить выполнение после N событий.

6.6 Логика фильтрации

Если событие остановило выполнение:

Определяется сегмент расходимости
Применяются фильтры (savitzky_golay, kolmogorov_zurbenko)
Вычисляется среднее отфильтрованных значений
Результат сохраняется в filtered_results

Результат: list[TrialResult]

Шаг 7. Параллельное/последовательное выполнение

Файлы: infra/trials/parallel_runner.py, infra/trials/sequential_runner.py

for result_chunk in runner.run(combinations):
    results.extend(result_chunk)

SequentialTrialRunner:

Выполняет комбинации последовательно
Yield’ит результаты по мере готовности

ParallelTrialRunner:

Создаёт pool = multiprocessing.Pool(process_count)
Распределяет комбинации по процессам
Yield’ит результаты по мере готовности

Результат: Iterator[list[TrialResult]]

Шаг 8. Сериализация результатов

Файл: infra/export/serializer.py

dicts = serializer.to_dict(results)

Преобразование TrialResult -> dict:

Все поля flatten в плоскую структуру
Числовые типы преобразуются в JSON-совместимые
Сложные объекты сериализуются

Шаг 9. Экспорт результатов

Файлы: infra/export/*.py

for exporter in exporters:
    exporter.export(dicts, config, series)

JSON Exporter

Файл: json_exporter.py

# Сохраняет в: output/results.json
json.dump(dicts, file, indent=2)

CSV Exporter

Файл: csv_exporter.py

# Сохраняет в: output/results.csv
df = pd.DataFrame(dicts)
df.to_csv(path, index=False)

–

Шаг 10. Визуализация (Frontend)

Web UI

React UI -> GET /api/results
FastAPI -> MongoDB с фильтрацией
MongoDB -> возвращает документы
FastAPI -> ResultDocument Pydantic модели
React UI -> отображает таблицы и графики

Vizr

PYTHONPATH="." python3 -m vizr.main ../runner/output/results.json

Загружает отфильтрованные JSON файл через Polars
Строит графики сходимости через PyQtGraph

Данные и их трансформация

Входные данные

Источник	Формат	Содержимое
`options.json`	JSON	Конфигурация выполнения
`example.json`	JSON	Ряды, методы, шумы, фильтры
`example_series.csv`	CSV	Натуральные ряды

Промежуточные данные

Стадия	Тип	Описание
Параметры	`BaseSeriesParam`, `BaseAccelParam`	Загруженные параметры
Комбинации	`tuple[Series, Accel]`	Декартово произведение
Trial result	`TrialResult`	Результат одного trial
Сериализованные	`dict`	JSON-совместимые словари

Выходные данные

Формат	Расположение	Использование
JSON	`output/results.json`	Анализ, интеграция
CSV	`output/results.csv`	Excel, pandas
Parquet	`output/results.parquet`	Vizr, big data
MongoDB	База данных	Web UI

Схема данных TrialResult

TrialResult
├── SeriesTrialResult
│   ├── id: str
│   ├── name: str
│   ├── lim: float | None
│   └── arguments: dict
├── AccelTrialResult
│   ├── name: str
│   ├── m_value: int
│   └── additional_args: dict
├── computed: list[ComputedTrialResult]
│   ├── n: int
│   ├── series_value: float
│   ├── partial_sum: float
│   ├── partial_sum_deviation: float
│   ├── accel_value: float
│   ├── accel_value_deviation: float
│   └── events: list[Event]
├── noise: NoiseConfig | None
├── error: ErrorTrialResult | NoErrorTrialResult
└── filtered: FilteredResults | None
    ├── start_n: int
    ├── segment_length: int
    └── methods: dict[str, FilterMethodResult]