def read_constituent_at_date(dirpath, date):
    date_csv = '%s.csv' % date.strftime('%Y-%m-%d')

    path = os.path.join(dirpath, date_csv)
    df = read_constituent(path)
    if df is None:
        # 对于指数数据，长期在线
        date_csv = '0.csv'
        path = os.path.join(dirpath, date_csv)
        df = read_constituent(path)
    return df def merge_constituent_date(constituent, ipo_last_trade, first_last):
    constituent_dt = constituent.merge(ipo_last_trade, how='left', left_on='wind_code', right_on='wind_code')
    if first_last is None:
        constituent_dt['first'] = pd.NaT
        constituent_dt['last'] = pd.NaT
    else:
        constituent_dt = constituent_dt.merge(first_last, how='left', left_on='wind_code', right_index=True)

    constituent_dt['start'] = constituent_dt.apply(lambda row: max(row['ipo_date'], row['last']), axis=1)
    constituent_dt['end'] = constituent_dt.apply(lambda row: min(row['lasttrade_date'], datetime.today()), axis=1)
    # 没有结束时间的，调整成当前时间
    constituent_dt['end'] = constituent_dt['end'].fillna(datetime.today())

    if True:
        # 不下载仿真合约
        constituent_dt = constituent_dt[~constituent_dt['sec_name'].str.contains('仿真')]

    return constituent_dt __CONFIG_TDAYS_SHFE_FILE__ = os.path.join(__CONFIG_H5_FUT_DIR__, 'tdays', 'SHFE.csv') __CONFIG_H5_FUT_SECTOR_DIR__ = os.path.join(__CONFIG_H5_FUT_DIR__, 'sectorconstituent') def split_alpha_number(symbol):
    """
    分离字母开头，然后接数字的情况，如果后面有别的字符会被截断
    print(split_alpha_number("TF1703.CFE"))
    ('TF', '1703')
    :param symbol:
    :return:
    """
    pattern = re.compile(r'([A-Za-z]+)(\d+)')
    match = pattern.match(symbol)
    if match:
        return match.group(1, 2)
    return None def yyyyMMdd_2_datetime(dt):
    yyyyMMdd = int(dt)
    (yyyy, MMdd) = divmod(yyyyMMdd, 10000)
    (MM, dd) = divmod(MMdd, 100)
    if yyyy == 0:
        return None

    return datetime(yyyy, MM, dd, 0, 0) def datetime_2_yyyyMMddHHmm(dt):
    """
    将时间转换成float类型
    :param dt:
    :return:
    """
    t = dt.timetuple()
    return float((t.tm_year * 10000.0 + t.tm_mon * 100 + t.tm_mday) * 10000.0) + t.tm_hour * 100 + t.tm_min def read_tdays(path):
    try:
        df = pd.read_csv(path, parse_dates=True)
    except:
        return None

    df['date'] = pd.to_datetime(df['date'])
    df.index = df['date']
    return df def download_min_ohlcv(w, wind_code, start, end):
    """
    下载分钟数据
    :param w:
    :param wind_code:
    :param start:
    :param end:
    :param exchange:
    :return:
    """
    # 在分数据数上不能改写时间函数，否则返回的时间有错
    # w.asDateTime = asDateTime
    fields = 'open,high,low,close,volume,amt,oi'
    w_wsd_data = w.wsi(wind_code, fields, start, end, '')
    df = pd.DataFrame(w_wsd_data.Data, )
    df = df.T
    df.columns = w_wsd_data.Fields
    df.index = w_wsd_data.Times
    return df, w_wsd_data.Codes[0] def read_datetime_dataframe(path, sep=','):
    """
    读取季报的公告日，即里面全是日期
    注意：有些股票多个季度一起发，一般是公司出问题了，特别是600878，四个报告同一天发布
    年报与一季报很有可能一起发

    读取ipo_date
    :param path:
    :param sep:
    :return:
    """
    try:
        df = pd.read_csv(path, index_col=0, parse_dates=True, encoding='utf-8-sig', sep=sep)
    except (FileNotFoundError, OSError):
        return None

    df = df.stack(dropna=False)
    df = pd.to_datetime(df)
    df = df.unstack()

    return df
import os
import sys
import pandas as pd
from datetime import datetime, timedelta
from WindPy import w
from demo_future.E01_download_daily import read_constituent_at_date, merge_constituent_date
from kquant_data.config import __CONFIG_TDAYS_SHFE_FILE__, __CONFIG_H5_FUT_SECTOR_DIR__
from kquant_data.utils.symbol import split_alpha_number
from kquant_data.utils.xdatetime import yyyyMMdd_2_datetime, datetime_2_yyyyMMddHHmm
from kquant_data.wind.tdays import read_tdays
from kquant_data.wind.wsi import download_min_ohlcv
from kquant_data.xio.csv import read_datetime_dataframe
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
下载合约分钟
由于数据量很大，可以只下载一些关键的
比如
主力、次主力
近月、远月

一定要通过update_first_last.py更新数据的下载范围
否则会全下，还会覆盖前面的数据
"""


# 解决Python 3.6的pandas不支持中文路径的问题
print(sys.getfilesystemencoding())  # 查看修改前的
try:
    sys._enablelegacywindowsfsencoding()  # 修改
    print(sys.getfilesystemencoding())  # 查看修改后的
except:
    pass


def download_constituent_min(w, dirpath, date, ipo_last_trade, first_last, wind_code_set, trading_days):
    constituent = read_constituent_at_date(dirpath, date)
    if constituent is None:
        # 没有对应的板块，因当是与上次一样导致
        # 没关系，上次数据应当已经下载过了
        return wind_code_set
    constituent_dt = merge_constituent_date(constituent, ipo_last_trade, first_last)

    for i in range(constituent_dt.shape[0]):
        row = constituent_dt.iloc[i]
        wind_code = row['wind_code']
        # 当前会话，不重复下载
        if wind_code in wind_code_set:
            continue
        wind_code_set.add(wind_code)
        # 时间已经到期了，不重复下载
        # 这里考虑时间
        if datetime_2_yyyyMMddHHmm(row['start']) == datetime_2_yyyyMMddHHmm(row['end']):
            continue

        product, num = split_alpha_number(wind_code)
        path_dir = os.path.join(root_path, product)
        if not os.path.exists(path_dir):
            os.mkdir(path_dir)
        path_csv = os.path.join(path_dir, '%s.csv' % wind_code)

        # 从开始到结束，可能跨度太长，特别是新下数据，可能超过一年，所以决定按月下载，这样更快
        # 保存时如果多次保存，然后放在一个文件中，会有问题
        trading_days['idx'] = range(len(trading_days))

        start = row['start']
        end = row['end']
        trading_days_idx = trading_days[start._date_repr:end._date_repr]['idx']
        rng = list(range(trading_days_idx[0], trading_days_idx[-1], 30))
        # 右闭
        rng.insert(len(rng), trading_days_idx[-1])
        for idx, r in enumerate(rng):
            if idx == 0:
                continue
            start_ = trading_days.iloc[rng[idx - 1]]['date']
            end_ = trading_days.iloc[r]['date']
            if idx == 1:
                # 第一个位置比较特殊，要取到前一个交易日的晚上
                start_ = trading_days.iloc[rng[idx - 1] - 1]['date']

            start_ += pd.Timedelta('20H')  # 会遇到不是交易日的问题
            end_ += pd.Timedelta('16H')
            print(start_, end_)
            try:
                df_old = pd.read_csv(path_csv, index_col=0, parse_dates=True)
                # 合并前先删除空数据
                df_old.dropna(axis=0, how='all', thresh=3, inplace=True)
            except:
                df_old = None
            print(row)
            df_new, wind_code = download_min_ohlcv(w, wind_code, start, end)
            df = pd.concat([df_old, df_new])
            df = df[~df.index.duplicated(keep='last')]
            print(path_csv)
            df.to_csv(path_csv)
        # 只做一个测试
        # break

    return wind_code_set


if __name__ == '__main__':
    # 数据太多，是否减少数据下载更合适？比如只下主力，次主力，最近月，最远月
    # 也可选择全下

    # 分钟数据是按天分割，还是放在一个文件中呢？
    # 下载是从哪下到哪呢？

    w.start()

    # 注意，数据中有可能出现0
    path_ipo_last_trade = os.path.join(__CONFIG_H5_FUT_SECTOR_DIR__, 'ipo_last_trade_trading.csv')
    ipo_last_trade = pd.read_csv(path_ipo_last_trade)
    ipo_last_trade['ipo_date'] = ipo_last_trade['ipo_date'].apply(lambda x: yyyyMMdd_2_datetime(x))
    ipo_last_trade['lasttrade_date'] = ipo_last_trade['lasttrade_date'].apply(lambda x: yyyyMMdd_2_datetime(x))

    root_path = r'D:\DATA_FUT\60'
    path_first_last = os.path.join(root_path, 'first_last.csv')
    first_last = read_datetime_dataframe(path_first_last)

    date_str = (datetime.today() - timedelta(days=1)).strftime('%Y-%m-%d')
    # 设置至少30天比较靠谱
    start_str = (datetime.today() - timedelta(days=30)).strftime('%Y-%m-%d')


    trading_days = read_tdays(__CONFIG_TDAYS_SHFE_FILE__)