Spaces:
Running
Running
File size: 3,313 Bytes
0f10134 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
import os
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
# 读取配置
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///data/stock_analyzer.db')
USE_DATABASE = os.getenv('USE_DATABASE', 'False').lower() == 'true'
# 创建引擎
engine = create_engine(DATABASE_URL)
Base = declarative_base()
# 定义模型
class StockInfo(Base):
__tablename__ = 'stock_info'
id = Column(Integer, primary_key=True)
stock_code = Column(String(10), nullable=False, index=True)
stock_name = Column(String(50))
market_type = Column(String(5))
industry = Column(String(50))
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
def to_dict(self):
return {
'stock_code': self.stock_code,
'stock_name': self.stock_name,
'market_type': self.market_type,
'industry': self.industry,
'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S') if self.updated_at else None
}
class AnalysisResult(Base):
__tablename__ = 'analysis_results'
id = Column(Integer, primary_key=True)
stock_code = Column(String(10), nullable=False, index=True)
market_type = Column(String(5))
analysis_date = Column(DateTime, default=datetime.now)
score = Column(Float)
recommendation = Column(String(100))
technical_data = Column(JSON)
fundamental_data = Column(JSON)
capital_flow_data = Column(JSON)
ai_analysis = Column(Text)
def to_dict(self):
return {
'stock_code': self.stock_code,
'market_type': self.market_type,
'analysis_date': self.analysis_date.strftime('%Y-%m-%d %H:%M:%S') if self.analysis_date else None,
'score': self.score,
'recommendation': self.recommendation,
'technical_data': self.technical_data,
'fundamental_data': self.fundamental_data,
'capital_flow_data': self.capital_flow_data,
'ai_analysis': self.ai_analysis
}
class Portfolio(Base):
__tablename__ = 'portfolios'
id = Column(Integer, primary_key=True)
user_id = Column(String(50), nullable=False, index=True)
name = Column(String(100))
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
stocks = Column(JSON) # 存储股票列表的JSON
def to_dict(self):
return {
'id': self.id,
'user_id': self.user_id,
'name': self.name,
'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else None,
'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S') if self.updated_at else None,
'stocks': self.stocks
}
# 创建会话工厂
Session = sessionmaker(bind=engine)
# 初始化数据库
def init_db():
Base.metadata.create_all(engine)
# 获取数据库会话
def get_session():
return Session()
# 如果启用数据库，则初始化
if USE_DATABASE:
init_db() |