"""Период консолидации: отчётная дата и выбор источников ОСВ (spec consolidations-api «Создание консолидации»; spec consolidated-osv «Выбор источников и приведение периодов»; design D-2; §2 `05-domain-rules.md`). Чистые функции без БД и ORM: документы приходят структурно (`DocLike` — `Document`, `SourceDocument`, тестовые dataclass'ы). ASSUMPTION(OQ-40): отчётная дата — первый день месяца, следующего за кварталом. """ import datetime import uuid from collections.abc import Sequence from dataclasses import dataclass from typing import Generic, Protocol, TypeVar from src.models.enums import PeriodQuarter __all__ = [ "QUARTER_ORDER", "DocLike", "Gap", "Sources", "counts_as_cumulative", "effective_cumulative", "quarters_through", "reporting_date", "select_sources", ] #: Порядок кварталов года — единственное определение (T8.5): чек-лист и привязка документов берут его отсюда. QUARTER_ORDER: tuple[PeriodQuarter, ...] = (PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3, PeriodQuarter.Q4) # квартал → (месяц отчётной даты, сдвиг года): Q4 заканчивается 31 декабря, отчётная дата — 1 января следующего года _REPORTING_MONTH: dict[PeriodQuarter, tuple[int, int]] = { PeriodQuarter.Q1: (4, 0), PeriodQuarter.Q2: (7, 0), PeriodQuarter.Q3: (10, 0), PeriodQuarter.Q4: (1, 1), } def reporting_date(year: int, quarter: PeriodQuarter) -> datetime.date: """Отчётная дата консолидации: Q1 → 01.04.year, Q2 → 01.07.year, Q3 → 01.10.year, Q4 → 01.01.year+1. ASSUMPTION(OQ-40): правило банка не подтверждено аналитикой (в ТЗ отчётная дата не определена). """ month, year_shift = _REPORTING_MONTH[quarter] # ASSUMPTION(OQ-40) return datetime.date(year + year_shift, month, 1) def quarters_through(quarter: PeriodQuarter) -> tuple[PeriodQuarter, ...]: """Кварталы года от `Q1` до `quarter` включительно (квартальные ОСВ, нужные версии за `quarter`).""" return QUARTER_ORDER[: QUARTER_ORDER.index(quarter) + 1] def counts_as_cumulative(is_cumulative: bool | None) -> bool: """Нарастающий ли документ: `is_cumulative` true или NULL (ББ/ОФР всегда с начала года, решение №10, R-1); квартальный — только явный `False`.""" return is_cumulative is not False def effective_cumulative(is_cumulative: bool | None, period_quarter: PeriodQuarter | None) -> bool: """Нарастающий ли документ для привязки, ключа замены и выбора квартальных источников (R-14). Документ за `Q1` равнозначен и нарастающему, и квартальному (нарастающий с начала года за первый квартал и есть квартал `Q1`; парсер помечает такие ОСВ нарастающими): здесь он считается квартальным при любом флаге — подходит любой версии года, а нарастающий и квартальный `Q1` одной пары — один слот привязки. Прочие — `counts_as_cumulative`. Не зависит от типа документа: ограничение «только ОСВ» (R-14a) живёт у вызывающего — `consolidation_documents._cumulative`; для ББ/ОФР эту функцию вызывать нельзя. """ return period_quarter != PeriodQuarter.Q1 and counts_as_cumulative(is_cumulative) class DocLike(Protocol): """Документ-кандидат в источники (структурный тип): поля — как в DBML `document`.""" @property def id(self) -> uuid.UUID: ... @property def period_quarter(self) -> PeriodQuarter | None: ... @property def is_cumulative(self) -> bool | None: ... D = TypeVar("D", bound=DocLike) @dataclass(frozen=True) class Sources(Generic[D]): """Источники пары (ЮЛ, счёт): нарастающий документ за квартал версии (`cumulative`) либо квартальные `Q1..Qn` в порядке кварталов.""" documents: tuple[D, ...] cumulative: bool @dataclass(frozen=True) class Gap: """Пробел пары (ЮЛ, счёт): нет квартальных ОСВ за `missing_quarters` (в порядке кварталов), строк по паре нет.""" missing_quarters: tuple[PeriodQuarter, ...] def select_sources(version_quarter: PeriodQuarter, documents: Sequence[D]) -> Sources[D] | Gap: """Выбор источников одной пары (ЮЛ, счёт) по документам, привязанным к версии (D-2, §2, OQ-03). 1. Нарастающий (`is_cumulative` true/NULL) за квартал версии есть → он один (`cumulative = True`), квартальные рядом игнорируются (нарастающий полнее по определению). Для версии `Q1` это нарастающий `Q1` «как есть». 2. Иначе нужны квартальные за каждый квартал `Q1..Qn` версии (`effective_cumulative` ложен: `is_cumulative = False` или любой документ за `Q1` — R-14, нарастающий `Q1` для версий `Q2+` служит квартальным `Q1`); для версии `Q1` квартальный `Q1` равнозначен нарастающему. Все есть → `Sources(cumulative=False)` в порядке кварталов (внутри квартала — порядок входа). 3. Какого-то квартала нет → `Gap(missing_quarters)`; при отсутствии подходящих документов — все кварталы `Q1..Qn`. Документы без квартала, нарастающие за другой квартал и квартальные позже версии источниками не бывают (такие документы не привязываются, D-5; функция их защитно игнорирует). Дублей на квартал привязка не допускает (R-1): если они всё же пришли, возвращаются все. """ cumulative = tuple( d for d in documents if counts_as_cumulative(d.is_cumulative) and d.period_quarter == version_quarter ) if cumulative: return Sources(documents=cumulative, cumulative=True) needed = quarters_through(version_quarter) quarterly = [ d for d in documents if not effective_cumulative(d.is_cumulative, d.period_quarter) and d.period_quarter in needed ] missing = tuple(quarter for quarter in needed if all(d.period_quarter != quarter for d in quarterly)) if missing: return Gap(missing_quarters=missing) ordered = tuple(d for quarter in needed for d in quarterly if d.period_quarter == quarter) return Sources(documents=ordered, cumulative=False)