From 07e07dfbcd48d1259773541f789adbe375f8a1b7 Mon Sep 17 00:00:00 2001 From: Ayuriel Date: Fri, 14 Mar 2025 18:18:39 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=A9=80=ED=8B=B0=ED=8C=A9=ED=84=B0=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- example/13-multi-factor-portfolio.py | 2 + quantcommon.py | 15 +- streamlit-quant/strategy/multi-factor.py | 262 +++++++++++++++++++++++ 3 files changed, 278 insertions(+), 1 deletion(-) create mode 100644 streamlit-quant/strategy/multi-factor.py diff --git a/example/13-multi-factor-portfolio.py b/example/13-multi-factor-portfolio.py index 39d977c..d04f3ed 100644 --- a/example/13-multi-factor-portfolio.py +++ b/example/13-multi-factor-portfolio.py @@ -6,6 +6,8 @@ import matplotlib.pyplot as plt import seaborn as sns import quantcommon +# strategy/multi-factor에서 구현 + # 멀티 팩터 포트폴리오. # 퀄리티: 자기자본이익률(ROE), 매출총이익(GPA), 영업활동현금흐름(CFO) # 밸류: PER, PBR, PSR, PCR, DY diff --git a/quantcommon.py b/quantcommon.py index 2cc3ee3..8b3f5b7 100644 --- a/quantcommon.py +++ b/quantcommon.py @@ -96,4 +96,17 @@ class QuantCommon: finally: engine.dispose() - return fs_list \ No newline at end of file + return fs_list + + def get_sector_list(self): + engine = self.create_engine() + + try: + sector_list = pd.read_sql(""" + select * from kor_sector + where 기준일 = (select max(기준일) from kor_sector); + """, con=engine) + finally: + engine.dispose() + + return sector_list \ No newline at end of file diff --git a/streamlit-quant/strategy/multi-factor.py b/streamlit-quant/strategy/multi-factor.py new file mode 100644 index 0000000..e5a3c57 --- /dev/null +++ b/streamlit-quant/strategy/multi-factor.py @@ -0,0 +1,262 @@ +import pandas as pd +import numpy as np +import statsmodels.api as sm +from scipy.stats import zscore +import matplotlib.pyplot as plt +import seaborn as sns +import quantcommon + +# 멀티 팩터 포트폴리오. +# 퀄리티: 자기자본이익률(ROE), 매출총이익(GPA), 영업활동현금흐름(CFO) +# 밸류: PER, PBR, PSR, PCR, DY +# 모멘텀: 12개월 수익률, K-Ratio +engine = quantcommon.QuantCommon().create_engine() + +# 각 섹터별 아웃라이어를 제거한 후 순위와 z-score를 구하는 함수 +def col_clean(df, cutoff=0.01, asc=False): + + q_low = df.quantile(cutoff) + q_hi = df.quantile(1 - cutoff) + + # 이상치 데이터 제거 + df_trim = df[(df > q_low) & (df < q_hi)] + + df_z_score = df_trim.rank(axis=0, ascending=asc).apply( + zscore, nan_policy='omit') + + return df_z_score + + +def plot_rank(df): + ax = sns.relplot(data=df, + x='rank', + y=1, + col='variable', + hue='invest', + size='size', + sizes=(10, 100), + style='invest', + markers={'Y': 'X','N': 'o'}, + palette={'Y': 'red','N': 'grey'}, + kind='scatter') + ax.set(xlabel=None) + ax.set(ylabel=None) + + plt.show() + + +def get_multi_factor_top(count): + qc = quantcommon.QuantCommon() + ticker_list = qc.get_ticker_list() + fs_list = qc.get_fs_list() + value_list = qc.get_value_list() + price_list = qc.get_price_list(12) + sector_list = qc.get_sector_list() + + # 퀄리티 지표 계산 + fs_list = fs_list.sort_values(['종목코드', '계정', '기준일']) + fs_list['ttm'] = fs_list.groupby(['종목코드', '계정'], as_index=False)['값'].rolling( + window=4, min_periods=4).sum()['값'] + fs_list_clean = fs_list.copy() + fs_list_clean['ttm'] = np.where(fs_list_clean['계정'].isin(['자산', '자본']), + fs_list_clean['ttm'] / 4, fs_list_clean['ttm']) + fs_list_clean = fs_list_clean.groupby(['종목코드', '계정']).tail(1) + + fs_list_pivot = fs_list_clean.pivot(index='종목코드', columns='계정', values='ttm') + fs_list_pivot['ROE'] = fs_list_pivot['당기순이익'] / fs_list_pivot['자본'] + fs_list_pivot['GPA'] = fs_list_pivot['매출총이익'] / fs_list_pivot['자산'] + fs_list_pivot['CFO'] = fs_list_pivot['영업활동으로인한현금흐름'] / fs_list_pivot['자산'] + + fs_list_pivot.round(4).head() + + value_list.loc[value_list['값'] <= 0, '값'] = np.nan + value_pivot = value_list.pivot(index='종목코드', columns='지표', values='값') + + value_pivot.head() + + # 가치 지표 계산 + # 음수를 제거하고 행으로 긴 형태로 변경 + price_pivot = price_list.pivot(index='날짜', columns='종목코드', values='종가') + ret_list = pd.DataFrame(data=(price_pivot.iloc[-1] / price_pivot.iloc[0]) - 1, + columns=['12M']) + + ret = price_pivot.pct_change().iloc[1:] + ret_cum = np.log(1 + ret).cumsum() + + x = np.array(range(len(ret))) + k_ratio = {} + + for i in range(0, len(ticker_list)): + ticker = ticker_list.loc[i, '종목코드'] + + try: + y = ret_cum.loc[:, price_pivot.columns == ticker] + reg = sm.OLS(y, x).fit() + res = float(reg.params / reg.bse) + except: + res = np.nan + + k_ratio[ticker] = res + + k_ratio_bind = pd.DataFrame.from_dict(k_ratio, orient='index').reset_index() + k_ratio_bind.columns = ['종목코드', 'K_ratio'] + + k_ratio_bind.head() + + # 가격 테이블을 이용해서 최근 12개월 수익률을 구하고 + # 로그 누적 수익률을 통해 각 종목별 K-Ratio를 계산 + data_bind = ticker_list[['종목코드', '종목명']].merge( + sector_list[['CMP_CD', 'SEC_NM_KOR']], + how='left', + left_on='종목코드', + right_on='CMP_CD').merge( + fs_list_pivot[['ROE', 'GPA', 'CFO']], how='left', + on='종목코드').merge(value_pivot, how='left', + on='종목코드').merge(ret_list, how='left', + on='종목코드').merge(k_ratio_bind, + how='left', + on='종목코드') + + data_bind.loc[data_bind['SEC_NM_KOR'].isnull(), 'SEC_NM_KOR'] = '기타' + data_bind = data_bind.drop(['CMP_CD'], axis=1) + + data_bind.round(4).head() + + # 종목코드와 섹터정보(SEC_NM_KOR)를 인덱스로 설정한 후, 섹터에 따른 그룹을 묶어준다. + data_bind_group = data_bind.set_index(['종목코드', + 'SEC_NM_KOR']).groupby('SEC_NM_KOR', as_index=False) + + data_bind_group.head(1).round(4) + + # 퀄리티 지표의 z-score를 계산 + # 퀄리티 지표에 해당하는 열(ROE, GPA, CFO) 선택해서 col_clean() 적용한 후 순위의 z-score 계산 + # sum() 함수를 통해 z-score의 합을 구하며, to_frame() 메소드를 통해 데이터프레임 형태로 변경 + z_quality = data_bind_group[['ROE', 'GPA', 'CFO' + ]].apply(lambda x: col_clean(x, 0.01, False)).sum( + axis=1, skipna=False).to_frame('z_quality') + # data_bind 테이블과 합치며, z_quality 열에는 퀄리티 지표의 z-score가 표시 + data_bind = data_bind.merge(z_quality, how='left', on=['종목코드', 'SEC_NM_KOR']) + + data_bind.round(4).head() + + # 가치 지표의 z-score 계산 + # 가치 지표에 해당하는 열(PBR, PCR, PER, PSR) 선택해서 col_clean() 적용, 오름차순 + value_1 = data_bind_group[['PBR', 'PCR', 'PER', + 'PSR']].apply(lambda x: col_clean(x, 0.01, True)) + # DY(배당수익률)의 경우 내림차순으로 계산 + value_2 = data_bind_group[['DY']].apply(lambda x: col_clean(x, 0.01, False)) + + # 두 결과를 합쳐 z-score의 합을 구한 후, 데이터프레임 형태로 변경 + z_value = value_1.merge(value_2, on=['종목코드', 'SEC_NM_KOR' + ]).sum(axis=1, + skipna=False).to_frame('z_value') + # data_bind 테이블과 합치며, z_value 열에는 밸류 지표의 z-score가 표시 + data_bind = data_bind.merge(z_value, how='left', on=['종목코드', 'SEC_NM_KOR']) + + data_bind.round(4).head() + + # 모멘텀 지표의 z-score 계산 + # 모멘텀 지표에 해당하는 열(12M, K_ratio) 선택 후 col_clean() 적용 + z_momentum = data_bind_group[[ + '12M', 'K_ratio' + ]].apply(lambda x: col_clean(x, 0.01, False)).sum( + axis=1, skipna=False).to_frame('z_momentum') + # data_bind 테이블과 합치며, z_momentum 열에는 모멘텀 지표의 z-score가 표시 + data_bind = data_bind.merge(z_momentum, how='left', on=['종목코드', 'SEC_NM_KOR']) + + data_bind.round(4).head() + + # 팩터 분포를 시각화 + data_z = data_bind[['z_quality', 'z_value', 'z_momentum']].copy() + + fig, axes = plt.subplots(3, 1, figsize=(10, 6), sharex=True, sharey=True) + for n, ax in enumerate(axes.flatten()): + ax.hist(data_z.iloc[:, n]) + ax.set_title(data_z.columns[n], size=12) + fig.tight_layout() + # plt.show() + + # 팩터 분포가 동일하지 않으니 z-score를 다시 계산해서 분포의 넓이를 비슷하게 맞춤 + # 종목 코드와 각 팩터의 z-score만 선택한 후, 종목 코드를 인덱스로 설정 + # apply()를 통해서 z-score 다시 계산 + data_bind_final = data_bind[['종목코드', 'z_quality', 'z_value', 'z_momentum' + ]].set_index('종목코드').apply(zscore, + nan_policy='omit') + # 열 이름 설정 + data_bind_final.columns = ['quality', 'value', 'momentum'] + + plt.rc('font', family='Malgun Gothic') + plt.rc('axes', unicode_minus=False) + fig, axes = plt.subplots(3, 1, figsize=(10, 6), sharex=True, sharey=True) + for n, ax in enumerate(axes.flatten()): + ax.hist(data_bind_final.iloc[:, n]) + ax.set_title(data_bind_final.columns[n], size=12) + fig.tight_layout() + # plt.show() + + # 각 팩터간 상관 관계 확인 + mask = np.triu(data_bind_final.corr()) + fig, ax = plt.subplots(figsize=(10, 6)) + sns.heatmap(data_bind_final.corr(), + annot=True, + mask=mask, + annot_kws={"size": 16}, + vmin=0, + vmax=1, + center=0.5, + cmap='coolwarm', + square=True) + ax.invert_yaxis() + # plt.show() + + # 각 팩터간 상관관계가 매우 낮으며, 여러 팩터를 동시에 고려함으로써 분산효과를 기대할 수 있다. + # 이제 계산된 팩터들을 토대로 최종 포트폴리오를 구성해 보자. + # 각 팩터를 동일 비중으로 설정. 0.2, 0.4, 0.4 등 중요하다고 생각되는 팩터에 비중을 다르게도 지정할 수 있음 + wts = [0.3, 0.3, 0.3] + # 위에서 설정한 비율을 반영해서 데이터프레임 형태로 변경 + data_bind_final_sum = (data_bind_final * wts).sum(axis=1, + skipna=False).to_frame() + data_bind_final_sum.columns = ['qvm'] + # 기본 테이블(data_bind)과 합침 + port_qvm = data_bind.merge(data_bind_final_sum, on='종목코드') + # 최종 z-score의 합(qvm) 기준 순위가 20위 이내인 경우 투자 종목에 해당하니 Y로 표시, 나머진 N + port_qvm['invest'] = np.where(port_qvm['qvm'].rank() <= 20, 'Y', 'N') + + # round()는 DataFrame 객체 내의 요소를 반올림하는 메서드 + return port_qvm[port_qvm['invest'] == 'Y'].round(4) + + # 이하 선택된 종목과 선택되지 않은 종목들 간의 특성을 그리기 위한 코드 + # data_melt = port_qvm.melt(id_vars='invest', + # value_vars=[ + # 'ROE', 'GPA', 'CFO', 'PER', 'PBR', 'PCR', 'PSR', + # 'DY', '12M', 'K_ratio' + # ]) + # + # data_melt['size'] = data_melt['invest'].map({'Y': 100, 'N': 10}) + # data_melt.head() + # + # hist_quality = data_melt[data_melt['variable'].isin(['ROE', 'GPA', + # 'CFO'])].copy() + # hist_quality['rank'] = hist_quality.groupby('variable')['value'].rank( + # ascending=False) + # plot_rank(hist_quality) + # + # hist_value = data_melt[data_melt['variable'].isin( + # ['PER', 'PBR', 'PCR', 'PSR', 'DY'])].copy() + # hist_value['value'] = np.where(hist_value['variable'] == 'DY', + # 1 / hist_value['value'], hist_value['value']) + # hist_value['rank'] = hist_value.groupby('variable')['value'].rank() + # plot_rank(hist_value) + # + # hist_momentum = data_melt[data_melt['variable'].isin(['12M', 'K_ratio'])].copy() + # hist_momentum['rank'] = hist_momentum.groupby('variable')['value'].rank(ascending = False) + # plot_rank(hist_momentum) + # + # port_qvm[port_qvm['invest'] == 'Y']['종목코드'].to_excel('model.xlsx', index=False) + + +if __name__ == '__main__': + top = get_multi_factor_top(20) + # head() 상위 n개 반환. 기본값은 5 + print(top.head()) + top['종목코드'].to_excel('model.xlsx', index=False) \ No newline at end of file