Add files via upload

Author: HaoningChen

scutquant/alpha.py

@@ -30,8 +30,7 @@ def market_neutralize(x: pd.Series, long_only: bool = False) -> pd.Series:
 
 def calc_factor_turnover(x: pd.Series) -> pd.Series:
     factor_neu = market_neutralize(x, long_only=False)
-    instrument_to = abs(factor_neu.groupby(level=1).diff())
-    instrument_to.dropna(inplace=True)
+    instrument_to = abs(factor_neu - ts_delay(factor_neu, 1).fillna(0))
     return instrument_to.groupby(level=0).sum()
 
 
@@ -97,7 +96,7 @@ def get_factor_metrics(factor: pd.Series, label: pd.Series, metrics=None, handle
     if "ir" in metrics:  # information ratio
         result["ir"] = result["excess_return"].mean() / result["return"].std()
     if "fitness" in metrics:
-        result["fitness"] = calc_fitness(result["sharpe"], result["return"].values[-1] - 1, result["turnover"].mean())
+        result["fitness"] = calc_fitness(result["sharpe"], result["return"].mean() - 1, result["turnover"].mean())
     return result
 
 
@@ -1075,10 +1074,10 @@ def call(self):
         volume_rank = cs_rank(self.data["volume"])
         rank_ratio = volume_rank / c_rank
         if isinstance(self.periods, int):
-            self.result = ts_decay_linear(-ts_rank(self.data["close"], self.periods) * rank_ratio, 15)
+            self.result = ts_decay(-ts_rank(self.data["close"], self.periods) * rank_ratio, 15)
         else:
             for d in self.periods:
-                self.result["wq1_" + str(d)] = ts_decay_linear(-ts_rank(self.data["close"], d) * rank_ratio, 15)
+                self.result["wq1_" + str(d)] = ts_decay(-ts_rank(self.data["close"], d) * rank_ratio, 15)
 
     def normalize(self):
         if self.norm_method == "zscore":
@@ -1115,7 +1114,7 @@ def __init__(self, data: pd.DataFrame, periods: list[int] | int, normalize: str
 
     def call(self):
         self.data["returns"] = ts_returns(self.data["close"], 1)
-        self.data["cs_mean"] = cs_mean(self.data["returns"]) * ts_decay_linear(self.data["close"], 15)
+        self.data["cs_mean"] = cs_mean(self.data["returns"]) * ts_decay(self.data["close"], 15)
         if isinstance(self.periods, int):
             self.result = cs_rank(ts_corr(self.data["returns"], self.data["cs_mean"], self.periods))
         else:

scutquant/data.py

@@ -1,216 +1,88 @@
-import akshare as ak
 import pandas as pd
-import datetime
+import tushare as ts
+import os
 
-# from joblib import Parallel, delayed
 
-"""
-akshare的数据并非100%准确！如果有更好的数据源请使用自己的数据
-不知为何sh000001和sh000002有问题
-"""
-
-
-def get_stock_data(instruments: list, freq: str = "daily", start: str = "19700101", end: str = "20230731",
-                   adjust: str = "") -> pd.DataFrame:
-    stock_data = pd.DataFrame()
-    for i in instruments:
-        single_stock = ak.stock_zh_a_hist(symbol=i[0: 6], period=freq, start_date=start, end_date=end, adjust=adjust)
-        single_stock["instrument"] = i
-        stock_data = pd.concat([stock_data, single_stock], axis=0)
-    stock_data.columns = ["datetime", "open", "close", "high", "low", "volume", "amount", "amplitude",
-                          "pct_chg", "price_chg", "turnover", "instrument"]
-    stock_data["datetime"] = pd.to_datetime(stock_data["datetime"])
-    stock_data.set_index(["datetime", "instrument"], inplace=True)
-    return stock_data.sort_index()
-
-
-def get_index_stock_cons(index_code='000300', freq="daily", start="20230330", end="20230331", adjust=""):
+def get_adj_hfq(price: pd.Series, pre_close: pd.Series) -> pd.Series:
     """
-    注：此函数还在不断完善中, 尤其是股票代码一块，非沪深300股票池的股票, 代码后缀可能会出错
-    :param index_code: str, 指数代码
-    :param freq: str, 有"daily", "weekly"和"monthly"可选
-    :param start: str, 日期, %y%m%d格式
-    :param end: str, 日期, %y%m%d格式
-    :param adjust: ""为不复权, “qfq”为前复权, “hfq”为后复权
-    :return: pd.DataFrame
-    example:
-    data = get_index_stock_cons()
+    计算后复权因子
     """
-    cons = ak.index_stock_cons(symbol=index_code)
-    df = pd.DataFrame()
-    for code in cons["品种代码"]:
-        stock_data = ak.stock_zh_a_hist(symbol=code, period=freq, start_date=start, end_date=end, adjust=adjust)
-        stock_data["instrument"] = code + ".SH" if code[0] == "6" else code + ".SZ"  # 根据股票代码的第一个数字区分其属于上交所还是深交所
-        df = pd.concat([df, stock_data], axis=0)
-    df = df.set_index(["日期", "instrument"]).sort_index()
-    df.index.names = ["datetime", "instrument"]
-    df = df[~df.index.duplicated()]
-    df.columns = ["open", "close", "high", "low", "volume", "amount", "amplitude", "price_chg", "pct_chg", "turnover"]
-    return df
+    price_ratio = (price / pre_close).groupby(level=1).transform(lambda x: x.cumprod())
+    adj = price_ratio.groupby(level=1).transform(lambda x: x / x[0])
+    return adj
 
 
-def upgrade_index_stock_cons(index_code='000300', today=None, adjust=""):
-    """
-    此函数设计的目的是自动更新数据
-    :param index_code: str, 指数代码
-    :param today: str, 今天的日期, %y%m%d格式
-    :param adjust: ""为不复权, “qfq”为前复权, “hfq”为后复权
-    :return: pd.DataFrame
-    example:
-    data = upgrade_index_stock_cons(today="20230330")
-    """
-    if today is None:
-        today = datetime.date.today()
-        today = today.strftime("%Y%m%d")
-    df = get_index_stock_cons(index_code=index_code, freq="daily", start=today, end=today, adjust=adjust)
-    return df
-
+def tus_init(tus_token: str = ""):
+    token = tus_token
+    ts.set_token(token)
+    pro = ts.pro_api()
+    return pro
 
-def get_daily_data(index_code, adjust=""):
-    """
-    获取指数成分股的历史数据(动态股票池, 日频), 支持各种复权
-    一次性获取所有日期的数据
-
-    :param index_code: 指数代码, like "sh000300"
-    :param adjust: ""为不复权, “qfq”为前复权, “hfq”为后复权
-    :return: pd.DataFrame
-    """
-    all_stocks = ak.index_stock_hist(symbol=index_code)
-    all_stocks["in_date"] = pd.to_datetime(all_stocks["in_date"]).dt.strftime('%Y%m%d')
-    all_stocks["out_date"] = pd.to_datetime(all_stocks["out_date"]).dt.strftime('%Y%m%d')
 
+def get_index_cons(pro, index_code: str = "000905.SH", start: str = "20100101", end: str = "20101231",
+                   output_folder: str = ""):
     data = pd.DataFrame()
-
-    for stock in all_stocks["stock_code"].unique():
-        start, end = all_stocks[all_stocks["stock_code"] == stock]["in_date"].unique(), \
-            all_stocks[all_stocks["stock_code"] == stock]["out_date"].unique()
-        # print(start, end)
-        for i in range(len(start)):
-            stock_data = ak.stock_zh_a_hist(symbol=stock, period="daily", start_date=start[i], end_date=end[i],
-                                            adjust=adjust)
-            stock_data["code"] = stock + ".SH" if stock[0] == "6" else stock + ".SZ"
-            data = pd.concat([data, stock_data], axis=0)
-    data = data.set_index(["日期", "code"]).sort_index()
-    data.index.names = ["datetime", "code"]
-    data = data[~data.index.duplicated()]
-    data.columns = ["open", "close", "high", "low", "volume", "amount", "amplitude", "price_chg", "pct_chg", "turnover"]
-    return data
-
-
-"""
-def get_high_freq_data(index_code="000300", minutes=1, adjust="hfq"):
-    def get_minute_data(code, minute, adj):
-        stock_code = "sh" + code if code[0] == "6" else "sz" + code
-        stock_data = ak.stock_zh_a_minute(symbol=stock_code, period=str(minute), adjust=adj)
-        stock_data["code"] = stock_code
-        stock_data.set_index(["day", "code"], inplace=True)
-        return stock_data
-
-    cons = ak.index_stock_cons(symbol=index_code)
-    df_list = Parallel(n_jobs=-1)(delayed(get_minute_data)(code, minutes, adjust) for code in cons["品种代码"])
-    df = pd.concat(df_list, axis=0)
-    df = df[~df.index.duplicated()]
-    return df
-"""
-
-
-def get_high_freq_data(index_code="000300", minutes=1, adjust="hfq"):
-    df = pd.DataFrame()
-    cons = ak.index_stock_cons(symbol=index_code)
-    for code in cons["品种代码"]:
-        stock_code = "sh" + code if code[0] == "6" else "sz" + code
-        stock_data = ak.stock_zh_a_minute(symbol=stock_code, period=str(minutes), adjust=adjust)
-        stock_data["code"] = stock_code
-        df = pd.concat([df, stock_data], axis=0)
-    df = df.set_index(["day", "code"]).sort_index()
-    df.dropna(axis=1, how='all', inplace=True)
-    df = df[~df.index.duplicated()]
-    return df
-
-
-"""
-# 并行计算会报错: no tables found
-def get_financial_data(index_code="000300", sleep=0.01):
-    def get_stock_data(code):
-        stock_data = ak.stock_financial_analysis_indicator(symbol=code)
-        stock_data["code"] = code + ".SH" if code[0] == "6" else code + ".SZ"
-        stock_data.set_index(["日期", "code"], inplace=True)
-        time.sleep(sleep)
-        return stock_data
-
-    cons = ak.index_stock_cons(symbol=index_code)
-    df_list = Parallel(n_jobs=-1)(delayed(get_stock_data)(code) for code in cons["品种代码"])
-    df = pd.concat(df_list, axis=0)
-    df.dropna(axis=1, how="all", inplace=True)
-    df.index.names = ["datetime", "code"]
-    df = df[~df.index.duplicated()]
-    return df
-"""
-
-
-def get_fundamental_data(index_code="000300"):
-    df = pd.DataFrame()
-    cons = ak.index_stock_cons(symbol=index_code)
-    for code in cons["品种代码"]:
-        stock_data = ak.stock_financial_analysis_indicator(symbol=code)
-        stock_data["code"] = code + ".SH" if code[0] == "6" else code + ".SZ"
-        df = pd.concat([df, stock_data], axis=0)
-    df = df.set_index(["日期", "code"]).sort_index()
-    df.dropna(axis=1, how="all", inplace=True)
-    df.index.names = ["datetime", "code"]
-    df = df[~df.index.duplicated()]
-    return df
-
-
-def get_futures_news(instrument="AL"):
-    """
-    由于期货是T0, 而新闻的datetime无法具体到分钟，而且新闻具有发布时间离散, 发布时集中(指同一天有多条新闻)的特点, 因此很难直接整合进行情数据中
-
-    :param instrument: 品种代码, 由于akshare采用的方法是代码后面+888(表示指数合约), 因此只要输入合约代码的前两位即可
-    :return: pd.DataFrame, 包括作为索引的datetime, instrument, 作为正式内容的新闻标题(akshare不返回正文内容)和正文链接
-
-    注: 链接点开会404, 所以没什么用
-
-    instrument 示例:
-    AL: 沪铝
-    J9: 焦炭
-    TA: PTA
-    CJ: 红枣
-    JM: 焦煤
-    """
-    news = ak.futures_news_baidu(symbol=instrument)
-    news.columns = ["title", "datetime", "link"]
-    news["instrument"] = instrument
-    return news.set_index(["datetime", "instrument"]).sort_index()
-
-
-def get_high_freq_futures(instrument="PTA", freq=1):
-    """
-    :param instrument: 资产名称, 品种大类的中文名, 例如PTA, 白糖等
-    :param freq: int, 频率, 1为1分钟, 以此类推
-    :return: pd.DataFrame
-    """
-    all_contracts = ak.futures_zh_realtime(symbol=instrument)["symbol"].tolist()
-    all_data = pd.DataFrame()
-    for contract in all_contracts:
-        data = ak.futures_zh_minute_sina(symbol=contract, period=str(freq))
-        data["instrument"] = contract
-        all_data = pd.concat([all_data, data], axis=0)
-    all_data.dropna(axis=1, how="all", inplace=True)
-    all_data.set_index(["datetime", "instrument"], inplace=True)
-    return all_data.sort_index()
-
-
-def get_stock_news(instrument_list: list) -> pd.DataFrame:
-    all_news = pd.DataFrame(())
-    for instrument in instrument_list:
-        i_news = ak.stock_news_em(instrument[0: 6])
-        i_news["关键词"] = instrument
-        # print(i_news)
-        all_news = pd.concat([all_news, i_news], axis=0)
-    all_news["发布时间"] = pd.to_datetime(all_news["发布时间"]).dt.strftime("%Y-%m-%d")
-    all_news["发布时间"] = pd.to_datetime(all_news["发布时间"])
-    all_news.set_index(["发布时间", "关键词"], inplace=True)
-    all_news.index.names = ["datetime", "instrument"]
-    all_news.columns = ["title", "content", "resource", "link"]
-    return all_news.sort_index()
+    data.index.names = ['datetime']
+    df = pd.DataFrame(pro.index_weight(index_code=index_code, start_date=start, end_date=end))  # 获得成分股列表
+    df.set_index(['trade_date'], inplace=True)
+    df.index.names = ['datetime']
+    df = df.sort_index()
+    data = pd.concat([data, df], axis=0).sort_index()
+    data.to_csv(output_folder + 'index_weight.csv')
+
+
+def process_index_cons(folder_path):
+    files = os.listdir(folder_path)
+    idx_cons = pd.DataFrame()
+
+    for file in files:
+        filepath = folder_path + file
+        sub_df = pd.read_csv(filepath)
+        sub_df.set_index("datetime", inplace=True)
+        code_list = pd.DataFrame()
+        codes = sub_df["con_code"].groupby(level=0).apply(lambda x: ','.join(x.astype(str)))
+        code_list["ts_code"] = codes
+        code_list["days"] = code_list.index.get_level_values(0)
+        code_list["days"] = code_list["days"].astype(str)
+        code_list["days"] = pd.to_datetime(code_list["days"], format="%Y-%m-%d")
+        # print(code_list)
+        code_list.reset_index(inplace=True)
+        code_list.set_index("days", inplace=True)
+        new_index = pd.date_range(start=code_list.index.min(), end=code_list.index.max(), freq='D')
+        code_list = code_list.reindex(new_index)
+        idx_cons = pd.concat([idx_cons, code_list], axis=0)
+    idx_cons.sort_index(inplace=True)
+    idx_cons.index.name = "days"
+    idx_cons["datetime"] = idx_cons.index.get_level_values(0).strftime("%Y%m%d").astype(int)
+    idx_cons.fillna(method="ffill", inplace=True)
+    idx_cons.to_csv("instrument_list.csv")
+
+
+def get_stock_data(pro, file_path='instrument_list.csv', adjust_price: bool = False) -> pd.DataFrame:
+    instrument_data = pd.DataFrame()
+    # 读取code_list后，按照list获取每支股票的数据
+    df1 = pd.read_csv(file_path)
+    df1.fillna(method='ffill', inplace=True)
+
+    date = df1['datetime'].unique()
+    day = []
+    for i in range(len(date)):
+        day.append(str(date[i]))
+
+    for i in range(len(date)):
+        df = pd.DataFrame(pro.daily(ts_code=str(df1['ts_code'].values[i]), start_date=day[i], end_date=day[i]))  # 行情数据
+        df['trade_date'] = pd.to_datetime(df['trade_date'])
+        df.set_index(['trade_date'], inplace=True)
+        df.index.names = ['datetime']
+        df = df.sort_index()
+        instrument_data = pd.concat([instrument_data, df], axis=0).sort_index()
+    instrument_data = instrument_data.reset_index()
+    instrument_data.set_index(["datetime", "ts_code"], inplace=True)
+    instrument_data.index.names = ["datetime", "instrument"]
+    if adjust_price:
+        adj = get_adj_hfq(instrument_data["close"], instrument_data["pre_close"])
+        # fixme: 增加调整volume的功能
+        prices = ["open", "close", "high", "low"]
+        for p in prices:
+            instrument_data[p] *= adj
+    return instrument_data

scutquant/operators.py

@@ -370,29 +370,20 @@ def ts_neg_count(data: pd.Series, n_period: int) -> pd.Series:
     return data_copy.groupby(level=1).transform(lambda x: x.rolling(n_period).sum())
 
 
-def linear_decay(x: pd.Series, window: int) -> pd.Series:
-    """
-    Applies linear decay to a time series.
-
-    :param x: The time series to apply linear decay to.
-    :type x: pd.Series
-    :param window: The window size for the linear decay.
-    :type window: int
-    :return: The time series with linear decay applied.
-    :rtype: pd.Series
-    """
-    weights = [np.exp(-1 / window * (window - t)) for t in range(window)]
-    return x.rolling(window).apply(lambda y: sum(y * weights) / sum(weights), raw=True)
+def decay_n(x: pd.Series, n: int) -> pd.Series:
+    arr = np.arange(1, n+1)
+    weights = arr / sum(arr)
+    return x.rolling(n).apply(lambda y: np.dot(y, weights), raw=True)
 
 
-def ts_decay_linear(data: pd.Series | pd.core.groupby.SeriesGroupBy, n_period: int) -> pd.Series:
+def ts_decay(data: pd.Series | pd.core.groupby.SeriesGroupBy, n_period: int) -> pd.Series:
     """
     Returns the linear decay on data for the past n_period days.
     """
     if isinstance(data, pd.Series):
-        return data.groupby(level=1).transform(lambda x: linear_decay(x, n_period))
+        return data.groupby(level=1).transform(lambda x: decay_n(x, n_period))
     else:
-        res: pd.Series = data.transform(lambda x: linear_decay(x, n_period))
+        res: pd.Series = data.transform(lambda x: decay_n(x, n_period))
         res.index.names = ["datetime", "instrument"]
         return res
 
@@ -599,9 +590,6 @@ def inf_mask(data: pd.Series) -> pd.Series:
 
 
 def get_resid(x: pd.Series, y: pd.Series) -> pd.Series:
-    """
-    经过百万级的数据的上千次实验, 发现此方法比调用sklearn.linear_model的LinearRegression平均快一倍
-    """
     cov = x.cov(y)
     var = x.var()
     beta = cov / var

scutquant/requirements.txt

@@ -1,4 +1,4 @@
-akshare>=1.9.59
+tushare>=1.2.8
 pandas>=1.5.3
 joblib>=1.2.0
 scipy>=1.10.0