Day 72：【99天精通Python】金融数据看板

Day 72：【99天精通Python】金融数据看板 - 数据层实现

前言

欢迎来到第72天！

在昨天的课程中，我们规划了项目的蓝图。今天，我们要开始打地基——构建数据层。
一个没有数据的看板就是个空壳。我们需要做两件事：

定义模型：告诉 Flask 数据库长什么样。
填充数据：编写脚本，从 BaoStock 接口把股票数据抓回来存进数据库。

本节内容：

配置 Flask-SQLAlchemy
编写models.py
编写数据抓取服务 (services/data_fetcher.py)
初始化数据库并测试抓取

一、配置 Flask 与数据库

在config.py中存放配置信息。

# config.pyimportos BASE_DIR=os.path.abspath(os.path.dirname(__file__))classConfig:# 数据库文件路径SQLALCHEMY_DATABASE_URI='sqlite:///'+os.path.join(BASE_DIR,'finance.db')# 关闭追踪修改，节省内存SQLALCHEMY_TRACK_MODIFICATIONS=False

在app.py中初始化 Flask 和 DB。

# app.pyfromflaskimportFlaskfromflask_sqlalchemyimportSQLAlchemyfromconfigimportConfig app=Flask(__name__)app.config.from_object(Config)# 初始化 SQLAlchemydb=SQLAlchemy(app)# 临时路由测试@app.route("/")defindex():return"Finance Board API"if__name__=="__main__":app.run(debug=True)

二、编写数据模型 (models.py)

根据昨天的设计，我们需要两个模型。

# models.pyfromappimportdbclassStockBasic(db.Model):__tablename__='stock_basic'# code 是主键，如 'sh.600519'code=db.Column(db.String(20),primary_key=True)code_name=db.Column(db.String(50))industry=db.Column(db.String(50))update_date=db.Column(db.Date)# 上次更新数据的日期def__repr__(self):returnf"<Stock{self.code_name}>"classStockDaily(db.Model):__tablename__='stock_daily'id=db.Column(db.Integer,primary_key=True)# 建立索引加快查询速度code=db.Column(db.String(20),index=True)date=db.Column(db.Date,index=True)open=db.Column(db.Float)high=db.Column(db.Float)low=db.Column(db.Float)close=db.Column(db.Float)volume=db.Column(db.Float)# 联合唯一索引：同一只股票同一天只能有一条数据__table_args__=(db.UniqueConstraint('code','date',name='unique_code_date'),)

初始化数据库：
打开终端进入项目目录：

python>>>from appimportdb, app>>>with app.app_context():... db.create_all()>>>exit()

此时目录下应该生成了finance.db。

三、编写数据抓取服务

我们需要一个独立的模块来处理与 BaoStock 的交互。

新建services/data_fetcher.py：

importbaostockasbsimportpandasaspdfromdatetimeimportdatetime,timedeltafromappimportdb,appfrommodelsimportStockBasic,StockDailyfromsqlalchemy.excimportIntegrityErrorclassDataFetcher:def__init__(self):self.system=bs.login()def__del__(self):bs.logout()deffetch_stock_list(self):"""获取所有 A 股列表并存入 StockBasic"""# 获取当天日期date=datetime.now().strftime("%Y-%m-%d")# query_all_stock 接口获取所有股票rs=bs.query_all_stock(day=date)data_list=[]whilers.next():data_list.append(rs.get_row_data())# BaoStock 返回: code, tradeStatus, code_nameforrowindata_list:code,status,name=row# 只存还在交易的股票ifstatus=='1'andcode.startswith(('sh','sz')):# 存入数据库stock=StockBasic.query.get(code)ifnotstock:stock=StockBasic(code=code,code_name=name)db.session.add(stock)db.session.commit()print(f"股票列表更新完成，共{len(data_list)}条")deffetch_daily_data(self,code,start_date=None,end_date=None):"""获取某只股票的日线数据"""ifnotend_date:end_date=datetime.now().strftime("%Y-%m-%d")ifnotstart_date:# 默认抓取过去 1 年start_date=(datetime.now()-timedelta(days=365)).strftime("%Y-%m-%d")rs=bs.query_history_k_data_plus(code,"date,open,high,low,close,volume",start_date=start_date,end_date=end_date,frequency="d",adjustflag="3")data_list=[]whilers.next():data_list.append(rs.get_row_data())# 批量入库forrowindata_list:# row: [date, open, high, low, close, volume]try:daily=StockDaily(code=code,date=datetime.strptime(row[0],"%Y-%m-%d").date(),open=float(row[1]),high=float(row[2]),low=float(row[3]),close=float(row[4]),volume=float(row[5])ifrow[5]else0)db.session.add(daily)exceptValueError:continue# 跳过无效数据try:db.session.commit()print(f"{code}数据导入成功，共{len(data_list)}条")exceptIntegrityError:db.session.rollback()print(f"{code}数据部分重复，已回滚")# 测试代码if__name__=="__main__":# 需要在 app context 下运行，因为用到了 dbwithapp.app_context():fetcher=DataFetcher()# 1. 更新股票列表 (跑一次就行)# fetcher.fetch_stock_list()# 2. 更新茅台数据fetcher.fetch_daily_data("sh.600519")

四、运行与验证

运行python services/data_fetcher.py。
观察控制台输出，确认数据下载成功。
使用 DB Browser 打开finance.db，查看stock_daily表，应该能看到一整年的茅台股价数据。

五、小结

关键要点：

ORM 模型：StockBasic存元数据，StockDaily存时间序列数据。
UniqueConstraint：数据库层面的约束，防止同一天插入两条重复数据。
App Context：在独立的脚本中使用 Flask 的db对象时，必须包裹在app.app_context()中。

六、课后作业

完善抓取逻辑：目前的fetch_stock_list比较慢，尝试只抓取我们感兴趣的几只股票（如上证50）。
断点更新：修改fetch_daily_data，每次抓取前先查询数据库，找出这只股票最新的一条数据的日期，然后只抓取那个日期之后的数据（增量更新）。
异常处理：如果在抓取过程中网络断了怎么办？给fetcher增加重试机制。