89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
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 PositionRanking(Base):
|
|
__tablename__ = "position_rankings"
|
|
__table_args__ = (UniqueConstraint("contract_code", "institution", "data_type", "date"),)
|
|
|
|
id: Mapped[int] = mapped_column(primary_key=True)
|
|
contract_code: Mapped[str] = mapped_column(String(10), index=True)
|
|
institution: Mapped[str] = mapped_column(String(30), index=True)
|
|
data_type: Mapped[str] = mapped_column(String(10), index=True)
|
|
date: Mapped[date] = mapped_column(Date, index=True)
|
|
rank: Mapped[int] = mapped_column(Integer)
|
|
value: Mapped[int] = mapped_column(Integer)
|
|
change: Mapped[int] = mapped_column(Integer)
|
|
|
|
|
|
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 |