CatB/backtrader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

♻️ refactor(data_fetcher): 将硬编码的表名提取为变量并统一管理，提高代码可维护性

Browse Source
This commit is contained in:
Qihang Zhang 2025-04-19 14:35:28 +08:00
parent 64de9058ff
commit 213df6aaff
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
data_fetcher.py
View File

@ -60,11 +60,12 @@ class DataFetcher:
"""
# 获取目标交易日历
all_trade_dates = self.get_trade_cal(start_date, end_date)
table_key = 'moneyflow_ind_dc'
# 确定需要获取的日期
if not force_update:
# 从数据库获取已有的交易日期
existing_dates = self.db_manager.get_existing_trade_dates(table_key='moneyflow_ind_dc')
existing_dates = self.db_manager.get_existing_trade_dates(table_key=table_key)
# 筛选出需要新获取的日期
dates_to_fetch = [date for date in all_trade_dates if date not in existing_dates]
else:
@ -74,7 +75,7 @@ class DataFetcher:
if not dates_to_fetch:
self.logger.info("所有数据已在数据库中，无需更新")
return self.db_manager.load_df_from_db(table_key='moneyflow_ind_dc')
return self.db_manager.load_df_from_db(table_key=table_key)
self.logger.info(f"需要获取 {len(dates_to_fetch)} 个交易日的数据")
@ -98,19 +99,19 @@ class DataFetcher:
new_df = pd.concat(all_new_data, ignore_index=True)
if force_update:
# 强制更新模式：需要删除已有的日期数据，然后重新插入
existing_df = self.db_manager.load_df_from_db(table_key='moneyflow_ind_dc')
existing_df = self.db_manager.load_df_from_db(table_key=table_key)
# 过滤掉需要更新的日期范围内的数据
filtered_df = existing_df[~existing_df['trade_date'].isin(dates_to_fetch)]
# 拼接新数据
final_df = pd.concat([filtered_df, new_df], ignore_index=True)
# 替换整个表
self.db_manager.save_df_to_db(final_df, table_key='moneyflow_ind_dc', if_exists='replace')
self.db_manager.save_df_to_db(final_df, table_key=table_key, if_exists='replace')
self.logger.info(f"已强制更新 {len(new_df)} 条记录到数据库")
else:
# 普通追加模式
self.db_manager.save_df_to_db(new_df, table_key='moneyflow_ind_dc', if_exists='append')
self.db_manager.save_df_to_db(new_df, table_key=table_key, if_exists='append')
self.logger.info(f"已将 {len(new_df)} 条新记录追加到数据库")
else:
self.logger.info("未获取到任何新数据")
return self.db_manager.load_df_from_db(table_key='moneyflow_ind_dc')
return self.db_manager.load_df_from_db(table_key=table_key)