import hashlib import os 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) class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(String(50), unique=True) password_hash: Mapped[str] = mapped_column(String(128)) role: Mapped[str] = mapped_column(String(20), default="user") @staticmethod def hash_password(password: str) -> str: salt = os.urandom(16) dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 100000) return salt.hex() + ":" + dk.hex() def check_password(self, password: str) -> bool: try: salt_hex, dk_hex = self.password_hash.split(":") salt = bytes.fromhex(salt_hex) dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 100000) return dk.hex() == dk_hex except (ValueError, AttributeError): return False