修正按周下载数据按年合并出错的问题

Author: wukan1986

ddump/_version.py

@@ -1 +1 @@
-__version__ = "0.1.5"
+__version__ = "0.1.6"

ddump/api/merge.py

@@ -44,6 +44,7 @@ def path_groupby_date(input_path, output_path,
     df = pd.DataFrame([f.name.split('.')[0].split(START_SEP_END) for f in files], columns=['start', 'end'])
     df['path'] = files
     df['key'] = pd.to_datetime(df['start'])
+    df['key2'] = df['key']
     df.index = df['key'].copy()
     df.index.name = 'date'  # 防止无法groupby
 
@@ -67,11 +68,15 @@ def path_groupby_date(input_path, output_path,
     df['10Y_1'] = df['key'].apply(lambda x: x.date() + relativedelta(year=x.year // 10 * 10, month=1, day=1))
     df['10Y_2'] = df['key'].apply(lambda x: x.date() + relativedelta(year=x.year // 10 * 10 + 9, month=12, day=31))
 
+    df['1M_1'] = pd.to_datetime(df['1M_1'])
+    df['1Y_1'] = pd.to_datetime(df['1Y_1'])
+
     # 最近的两个月不动，两个月前的都按月合并
     t = f'{datetime.now() - timedelta(days=31 * 2):%Y-%m}'
     df['key'] = df.loc[:t, '1M_1']
     t = f'{datetime.now() - timedelta(days=365 * 1):%Y}'
     df['key'] = df.loc[:t, '1Y_1']
+    df['key'].fillna(df['key2'], inplace=True)
 
     # 按key进行分组
     fss = {}

ddump/db/merge.py

@@ -110,6 +110,7 @@ def path_groupby_date(input_path, output_path,
     df = pd.DataFrame([f.name.split('.')[0].split(KEY_SEP_ID) for f in files], columns=['key', 'id'])
     df['path'] = files
     df['key'] = pd.to_datetime(df['key'])
+    df['key2'] = df['key']
     df.index = df['key'].copy()
     df.index.name = 'date'  # 防止无法groupby
 
@@ -133,11 +134,15 @@ def path_groupby_date(input_path, output_path,
     df['10Y_1'] = df['key'].apply(lambda x: x.date() + relativedelta(year=x.year // 10 * 10, month=1, day=1))
     df['10Y_2'] = df['key'].apply(lambda x: x.date() + relativedelta(year=x.year // 10 * 10 + 9, month=12, day=31))
 
+    df['1M_1'] = pd.to_datetime(df['1M_1'])
+    df['1Y_1'] = pd.to_datetime(df['1Y_1'])
+
     # 最近的两个月不动，两个月前的都按月合并
     t = f'{datetime.now() - timedelta(days=31 * 2):%Y-%m}'
     df['key'] = df.loc[:t, '1M_1']
     t = f'{datetime.now() - timedelta(days=365 * 1):%Y}'
     df['key'] = df.loc[:t, '1Y_1']
+    df['key'].fillna(df['key2'], inplace=True)
 
     # 按key进行分组
     fss = {}

ddump/merge.py

@@ -1,10 +1,13 @@
+import shutil
+
 import pandas as pd
 from loguru import logger
 
 
 def merge_files_to_file(path, files,
                         ignore_index=True,
-                        delete_src=False):
+                        delete_src=False,
+                        single_overwrite=True):
     """合并件列表到文件
 
     Parameters
@@ -17,17 +20,30 @@ def merge_files_to_file(path, files,
         合并时是否忽略索引。索引没有意义时忽略能加速
     delete_src: bool
         是否删除源文件
-
-    Returns
-    -------
+    single_overwrite: bool
+        单文件是否进行覆盖
 
     """
     if len(files) == 0:
         return
+
     if len(files) == 1:
         if path == files[0]:
             # 同一文件，没有必要合并
             return
+        else:
+            if path.exists() and not single_overwrite:
+                logger.info('单路径，已存在，跳过 {}', path)
+                return
+            else:
+                logger.info('单路径，直接覆盖 {}', path)
+                path.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copy(files[0], path)
+                return
+
+    if path.exists():
+        logger.info('合并目标，已存在，跳过 {}', path)
+        return
 
     # 加载
     dfs = []
@@ -69,5 +85,7 @@ def merge_files_dict(files_dict,
     key为路径
     value为列表
     """
-    for k, v in files_dict.items():
-        merge_files_to_file(k, v, ignore_index, delete_src)
+    for i, (k, v) in enumerate(files_dict.items()):
+        # 最后5个单文件总是试着覆盖
+        single_overwrite = i >= len(files_dict) - 5
+        merge_files_to_file(k, v, ignore_index, delete_src, single_overwrite)

examples/jqresearch/get_price_daily.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 
 import pandas as pd
+from dateutil.relativedelta import relativedelta
 
 from ddump.api.dump import Dump__start__end
 from examples.jqresearch.config import DATA_ROOT, jq
@@ -66,6 +67,7 @@ def do_get_industry(d, start_date, end_date, symbols):
     # 下载日线数据
     d.set_parameters('get_industry',
                      start_date=f'{start_date:%Y-%m-%d}',
+                     end_date=f'{end_date:%Y-%m-%d}',
                      date=f'{end_date:%Y-%m-%d}',
                      security=symbols.index.tolist())
     if not d.exists(file_timeout=3600 * 6, data_timeout=86400 * 2):
@@ -84,18 +86,31 @@ def do_get_industry(d, start_date, end_date, symbols):
     d1 = Dump__start__end(jq, path1, 'start_date', 'end_date')
     d2 = Dump__start__end(jq, path2, 'start_date', 'end_date')
     d3 = Dump__start__end(jq, path3, 'start_date', 'end_date')
-    d4 = Dump__start__end(jq, path4, 'start_date', 'date')
+    d4 = Dump__start__end(jq, path4, 'start_date', 'end_date')
 
     # 前半段，按周查，这样能快一些
     end = pd.to_datetime('2023-01-15')  # 星期日
     # 下周，由date_range调到本周日
     end = pd.to_datetime(datetime.today().date()) + pd.Timedelta(days=6)
     start = pd.to_datetime('2023-01-02')  # 星期一
-    start = pd.to_datetime('2015-01-01')  # 星期一
+
+    # 只要跨月了就划分成两部分，实现指定月份也能加载不出错
+    start_list = []
+    end_list = []
     for dr in pd.date_range(start=start, end=end, freq='W'):
         start_date = dr - pd.Timedelta(days=6)
         end_date = dr
-
+        if start_date.month == end_date.month:
+            start_list.append(start_date)
+            end_list.append(end_date)
+        else:
+            start_list.append(start_date)
+            end_list.append(start_date + relativedelta(day=31))
+            start_list.append(end_date + relativedelta(day=1))
+            end_list.append(end_date)
+
+    # 下载数据
+    for start_date, end_date in zip(start_list, end_list):
         symbols = universe.query(f'start_date<=@end_date.date() and end_date>=@start_date.date()')
 
         do_get_price(d1, start_date, end_date, symbols, fields1, fq1)

tests/merge_jqresearch.py renamed to examples/jqresearch/merge_jqresearch.py

@@ -1,12 +1,17 @@
 """
 下载历史数据后，很早以前的数据可以合并。一年合并一次即可
+
+历史数据如果还没有验证，应当选取其它方式验证多次后再合并
+因为只要发现某天数据有问题，只要删了那天的文件，即可重新下载对应部分
 """
 import pathlib
 
+from loguru import logger
+
 from ddump.api.merge import path_groupby_date
 from ddump.merge import merge_files_dict
 
-paths = [
+paths1 = [
     r'D:\data\jqresearch\get_extras_stock_is_st',
     r'D:\data\jqresearch\get_industry_stock',
     r'D:\data\jqresearch\get_price_stock_factor',
@@ -17,16 +22,21 @@
     r'D:\data\jqresearch\get_fundamentals_indicator',
     r'D:\data\jqresearch\get_fundamentals_valuation',
 ]
-for path in paths:
-    path = pathlib.Path(path)
-    files = path_groupby_date(path, path)
-    merge_files_dict(files, ignore_index=False, delete_src=True)
 
-if False:
-    # 测试用
-    path1 = r'D:\data\jqresearch\get_fundamentals_balance'
+paths2 = [
+    r'M:\data\jqresearch\get_extras_stock_is_st',
+    r'M:\data\jqresearch\get_industry_stock',
+    r'M:\data\jqresearch\get_price_stock_factor',
+    r'M:\data\jqresearch\get_price_stock_daily',
+    r'M:\data\jqresearch\get_fundamentals_balance',
+    r'M:\data\jqresearch\get_fundamentals_cash_flow',
+    r'M:\data\jqresearch\get_fundamentals_income',
+    r'M:\data\jqresearch\get_fundamentals_indicator',
+    r'M:\data\jqresearch\get_fundamentals_valuation',
+]
+for path1, path2 in zip(paths1, paths2):
+    logger.info('=' * 60, )
     path1 = pathlib.Path(path1)
-    path2 = r'D:\data\jqresearch\get_fundamentals_balance_2'
     path2 = pathlib.Path(path2)
     files = path_groupby_date(path1, path2)
     merge_files_dict(files, ignore_index=False, delete_src=False)