112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""история номинала, ручные цена/доходность/дюрация на дату
|
|
|
|
Revision ID: a4e81c6f2b97
|
|
Revises: 7c2e5a9d1f30
|
|
Create Date: 2026-10-21 00:00:00
|
|
|
|
security_nominal_history (D11), market_data_manual (D6).
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
import sqlalchemy as sa
|
|
|
|
from alembic import op
|
|
from app.core.config import settings
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = "a4e81c6f2b97"
|
|
down_revision: Union[str, None] = "7c2e5a9d1f30"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
_schema = settings.DATABASE_SCHEMA or None
|
|
|
|
# Типы зафиксированы в ревизии, как и в начальной миграции.
|
|
_BIGINT_PK = sa.BigInteger().with_variant(sa.Integer(), "sqlite")
|
|
_PRICE = sa.Numeric(18, 6)
|
|
_PERCENT = sa.Numeric(12, 6)
|
|
_QUANTITY = sa.Numeric(20, 4)
|
|
|
|
|
|
def _fk(table_column: str) -> str:
|
|
return f"{_schema}.{table_column}" if _schema else table_column
|
|
|
|
|
|
def _id() -> sa.Column:
|
|
return sa.Column("id", _BIGINT_PK, sa.Identity(), primary_key=True)
|
|
|
|
|
|
def _security_id() -> sa.Column:
|
|
return sa.Column(
|
|
"security_id",
|
|
sa.BigInteger(),
|
|
sa.ForeignKey(_fk("security.id")),
|
|
nullable=False,
|
|
)
|
|
|
|
|
|
def _audit() -> list[sa.Column]:
|
|
return [
|
|
sa.Column(
|
|
"created_at",
|
|
sa.DateTime(timezone=True),
|
|
server_default=sa.func.now(),
|
|
nullable=False,
|
|
),
|
|
sa.Column("created_by", sa.String(255), nullable=False),
|
|
sa.Column(
|
|
"updated_at",
|
|
sa.DateTime(timezone=True),
|
|
server_default=sa.func.now(),
|
|
nullable=False,
|
|
),
|
|
sa.Column("updated_by", sa.String(255), nullable=False),
|
|
]
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""Upgrade schema."""
|
|
op.create_table(
|
|
"security_nominal_history",
|
|
_id(),
|
|
_security_id(),
|
|
sa.Column("valid_from", sa.Date(), nullable=False),
|
|
sa.Column("face_value", _PRICE, nullable=False),
|
|
sa.Column("source", sa.String(10), nullable=False),
|
|
sa.Column("reason", sa.String(500), nullable=True),
|
|
*_audit(),
|
|
sa.UniqueConstraint(
|
|
"security_id",
|
|
"valid_from",
|
|
"source",
|
|
name="uq_security_nominal_history_date_source",
|
|
),
|
|
sa.CheckConstraint(
|
|
"source IN ('moex', 'manual')", name="ck_security_nominal_history_source"
|
|
),
|
|
schema=_schema,
|
|
)
|
|
|
|
op.create_table(
|
|
"market_data_manual",
|
|
_id(),
|
|
_security_id(),
|
|
sa.Column("market_date", sa.Date(), nullable=False),
|
|
sa.Column("clean_price", _PRICE, nullable=True),
|
|
sa.Column("yield_value", _PERCENT, nullable=True),
|
|
sa.Column("duration_days", _QUANTITY, nullable=True),
|
|
sa.Column("reason", sa.String(500), nullable=False),
|
|
*_audit(),
|
|
sa.UniqueConstraint(
|
|
"security_id", "market_date", name="uq_market_data_manual_security_date"
|
|
),
|
|
schema=_schema,
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""Downgrade schema."""
|
|
op.drop_table("market_data_manual", schema=_schema)
|
|
op.drop_table("security_nominal_history", schema=_schema)
|