from datetime import date from sqlalchemy import String, Integer, Float, Date, ForeignKey, Boolean, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) code: Mapped[str] = mapped_column(String(10), unique=True, index=True) name: Mapped[str] = mapped_column(String(20)) exchange: Mapped[str] = mapped_column(String(10), default="CZCE") contracts: Mapped[list["Contract"]] = relationship( back_populates="product", cascade="all, delete-orphan" ) class Contract(Base): __tablename__ = "contracts" __table_args__ = (UniqueConstraint("code"),) id: Mapped[int] = mapped_column(primary_key=True) product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), index=True) code: Mapped[str] = mapped_column(String(10), unique=True, index=True) name: Mapped[str] = mapped_column(String(30)) is_active: Mapped[bool] = mapped_column(Boolean, default=True) product: Mapped["Product"] = relationship(back_populates="contracts") class DailyBar(Base): __tablename__ = "daily_bars" __table_args__ = (UniqueConstraint("contract", "date"),) id: Mapped[int] = mapped_column(primary_key=True) contract: Mapped[str] = mapped_column(String(10), index=True) date: Mapped[date] = mapped_column(Date, index=True) open: Mapped[float] = mapped_column(Float) close: Mapped[float] = mapped_column(Float) high: Mapped[float] = mapped_column(Float) low: Mapped[float] = mapped_column(Float) amp_5d: Mapped[float | None] = mapped_column(Float, nullable=True) @property def diff(self) -> float: """最高-最低价差""" return self.high - self.low class PositionSnapshot(Base): __tablename__ = "position_snapshots" __table_args__ = (UniqueConstraint("institution", "direction", "date"),) id: Mapped[int] = mapped_column(primary_key=True) institution: Mapped[str] = mapped_column(String(20), index=True) direction: Mapped[str] = mapped_column(String(10)) # "long" or "short" date: Mapped[date] = mapped_column(Date, index=True) position: Mapped[int] = mapped_column(Integer) delta: Mapped[int] = mapped_column(Integer, default=0) avg_cost: Mapped[float] = mapped_column(Float)