feat: 슈퍼 퀄리티 전략 구현

src/data/prices.py

@@ -40,7 +40,7 @@ def process_for_price():
         # 시작일과 종료일
         # fr = (date.today() + relativedelta(years=-5)).strftime("%Y%m%d")
         to = (date.today()).strftime("%Y%m%d")
-        fr = '20250125'
+        fr = '20250301'
 
         # 오류 발생 시 이를 무시하고 다음 루프로 진행
         try:

src/strategies/composite/multi_factor.py

@@ -181,9 +181,9 @@ def get_multi_factor_top(qc, count):
     # 열 이름 설정
     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)
+    # 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)

src/strategies/composite/super_quality.py

@@ -1,6 +1,4 @@
 from datetime import datetime
-from src import ui as st
-
 from strategies.factors import f_score
 from db.common import DBManager
 
@@ -44,6 +42,14 @@ from db.common import DBManager
 # 신F-스코어가 3점인 종목 중 GP/A가 높은 종목 위주로 매수했으면 (1) CAGR도 조금 개선되고 (2) 최상 30개 종목을 매수했을 경우 선택받은 종목 360개 중 파산한 기업은 단 1개였다.
 # F-스코어와 GP/A는 엄청난 잠재력을 지닌 콤비네이션임이 분명하다.
 # """)
+def calc_gpa(db, base_date):
+    fs_list = db.get_fs_list_by_account_and_date("'매출총이익', '자산'", f"'{base_date}'")
+    fs_list_pivot = fs_list.pivot(index='종목코드', columns='계정', values='값')
+    fs_list_pivot['GPA'] = fs_list_pivot['매출총이익'] / fs_list_pivot['자산']
+
+    # 결과 정리
+    return fs_list_pivot.reset_index()[['종목코드', 'GPA']]
+
 
 def get_last_year_end():
     # 현재 날짜 가져오기 (2025년 3월 16일 기준)
@@ -57,11 +63,22 @@ def get_last_year_end():
 
     return last_year_end.date()
 
-st.write("투자 전략: 강환국 슈퍼 퀄리티 전략 2.0")
+def get_super_quality_top(db):
+    date = get_last_year_end()
+    f_score_data = f_score.get_f_score(DBManager(), date)
 
-date = get_last_year_end()
-data = f_score.get_f_score(DBManager(), date)
+    gpa_list = calc_gpa(db, date)
 
-config = {}
+    # GPA 병합 + NaN인 경우 기본 값 -1(내림차순 정렬 시에 하위 순위를 받게 하려고)
+    final_df = f_score_data.merge(gpa_list, on='종목코드', how='left')
+    final_df['GPA'] = final_df['GPA'].fillna(-1).astype(float)
 
-st.dataframe(data, column_config=config, use_container_width=True)
+    f_score3 = final_df[final_df['f-score'] == 3].round(4)
+    result = f_score3[f_score3['분류'] == '소형주'].sort_values('GPA', ascending=False)
+    # print(f_score3)
+    # fs_list_copy = f_score3[['GPA']].copy()
+    # # print(fs_list_copy)
+    # fs_rank = fs_list_copy.rank(ascending=False, axis=0)
+    # # print(fs_rank)
+    # return f_score3.loc[fs_rank['GPA'] <= 20, ['종목코드', '종목명', '분류', 'f-score', 'GPA']].round(4)
+    return result

src/strategies/factors/f_score.py

@@ -45,14 +45,6 @@ def calc_capital(qc, base_date):
     return pivot_df.reset_index()[['종목코드', 'score3']]
 
 
-def calc_gpa(qc, base_date):
-    fs_list = qc.get_fs_list_by_account_and_date("'매출총이익', '자산'", f"'{base_date}'")
-    fs_list_pivot = fs_list.pivot(index='종목코드', columns='계정', values='값')
-    fs_list_pivot['GPA'] = fs_list_pivot['매출총이익'] / fs_list_pivot['자산']
-
-    # 결과 정리
-    return fs_list_pivot.reset_index()[['종목코드', 'GPA']]
-
 def get_ticker_list(qc):
     ticker_list = qc.get_ticker_list()
     # 시가총액을 기준으로 정렬
@@ -67,7 +59,6 @@ def get_f_score(qc, base_date):
     score1_list = calc_net_income(qc, base_date)
     score2_list = calc_cfo(qc, base_date)
     score3_list = calc_capital(qc, base_date)
-    gpa_list = calc_gpa(qc, base_date)
 
     # score 1 병합 + NaN인 경우 기본값 0
     merge_score1 = ticker_list.merge(score1_list, on='종목코드', how='left')
@@ -84,20 +75,7 @@ def get_f_score(qc, base_date):
     # 개별 점수들로 신f-score 계산
     merge_score3['f-score'] = merge_score3['score1'] + merge_score3['score2'] + merge_score3['score3']
 
-    # GPA 병합 + NaN인 경우 기본 값 -1(내림차순 정렬 시에 하위 순위를 받게 하려고)
-    final_df = merge_score3.merge(gpa_list, on='종목코드', how='left')
-    final_df['GPA'] = final_df['GPA'].fillna(-1).astype(float)
-
-    f_score3 = final_df[final_df['f-score'] == 3].round(4)
-    result = f_score3[f_score3['분류'] == '소형주'].sort_values('GPA', ascending=False)
-
-    # print(f_score3)
-    # fs_list_copy = f_score3[['GPA']].copy()
-    # # print(fs_list_copy)
-    # fs_rank = fs_list_copy.rank(ascending=False, axis=0)
-    # # print(fs_rank)
-    # return f_score3.loc[fs_rank['GPA'] <= 20, ['종목코드', '종목명', '분류', 'f-score', 'GPA']].round(4)
-    return result
+    return merge_score3
 
 if __name__ == '__main__':
     date = datetime(2024, 12, 31).date()

src/ui/pages/quality_page.py

@@ -3,7 +3,7 @@ Super Quality strategy page for the Streamlit Quant application.
 """
 import streamlit as st
 from datetime import datetime
-from strategies.factors.f_score import get_f_score
+from strategies.composite.super_quality import get_super_quality_top
 from db.common import DBManager
 
 def render_quality_page():
@@ -60,59 +60,54 @@ def render_quality_page():
     st.write("## 슈퍼 퀄리티 전략 2.0 포트폴리오")
     
     # Get data
-    date = get_last_year_end()
     db = DBManager()
-    data = get_f_score(db, date)
-    
-    # Display options
-    col1, col2 = st.columns([1, 2])
-    
-    with col1:
-        st.write("### 설정")
-        min_f_score = st.slider("최소 F-스코어", min_value=0, max_value=3, value=3)
-        include_small_caps = st.checkbox("소형주만 포함", value=True)
-        num_stocks = st.slider("포트폴리오 종목수", min_value=5, max_value=50, value=20)
-    
-    # Filter data
-    filtered_data = data[data['f-score'] >= min_f_score].copy()
-    
-    if include_small_caps:
-        # Sort by market cap and keep only the bottom 20%
-        filtered_data = filtered_data.sort_values('시가총액')
-        filtered_data = filtered_data.head(int(len(filtered_data) * 0.2))
-    
-    # Sort by GP/A in descending order
-    filtered_data = filtered_data.sort_values('GP/A', ascending=False)
-    
-    # Get top N stocks
-    portfolio = filtered_data.head(num_stocks)
-    
-    # Display portfolio
-    with col2:
-        st.write(f"### 선택된 {len(portfolio)} 종목")
-        st.dataframe(portfolio, use_container_width=True)
-    
-    # Display metrics
-    st.write("### 포트폴리오 지표")
-    col1, col2, col3, col4 = st.columns(4)
-    
-    with col1:
-        st.metric(label="평균 F-스코어", value=f"{portfolio['f-score'].mean():.2f}")
-    
-    with col2:
-        st.metric(label="평균 GP/A", value=f"{portfolio['GP/A'].mean():.2f}%")
-    
-    with col3:
-        avg_market_cap = portfolio['시가총액'].mean() / 1_000_000_000
-        st.metric(label="평균 시가총액", value=f"{avg_market_cap:.1f}십억원")
-    
-    with col4:
-        st.metric(label="종목 수", value=len(portfolio))
+    data = get_super_quality_top(db)
 
+    st.write("### 포트폴리오")
+    st.write(data)
+    
+    # # Display options
+    # col1, col2 = st.columns([1, 2])
+    #
+    # with col1:
+    #     st.write("### 설정")
+    #     min_f_score = st.slider("최소 F-스코어", min_value=0, max_value=3, value=3)
+    #     include_small_caps = st.checkbox("소형주만 포함", value=True)
+    #     num_stocks = st.slider("포트폴리오 종목수", min_value=5, max_value=50, value=20)
+    #
+    # # Filter data
+    # filtered_data = data[data['f-score'] >= min_f_score].copy()
+    #
+    # if include_small_caps:
+    #     # Sort by market cap and keep only the bottom 20%
+    #     filtered_data = filtered_data.sort_values('시가총액')
+    #     filtered_data = filtered_data.head(int(len(filtered_data) * 0.2))
+    #
+    # # Sort by GP/A in descending order
+    # filtered_data = filtered_data.sort_values('GP/A', ascending=False)
+    #
+    # # Get top N stocks
+    # portfolio = filtered_data.head(num_stocks)
+    #
+    # # Display portfolio
+    # with col2:
+    #     st.write(f"### 선택된 {len(portfolio)} 종목")
+    #     st.dataframe(portfolio, use_container_width=True)
+    #
+    # # Display metrics
+    # st.write("### 포트폴리오 지표")
+    # col1, col2, col3, col4 = st.columns(4)
+    #
+    # with col1:
+    #     st.metric(label="평균 F-스코어", value=f"{portfolio['f-score'].mean():.2f}")
+    #
+    # with col2:
+    #     st.metric(label="평균 GP/A", value=f"{portfolio['GP/A'].mean():.2f}%")
+    #
+    # with col3:
+    #     avg_market_cap = portfolio['시가총액'].mean() / 1_000_000_000
+    #     st.metric(label="평균 시가총액", value=f"{avg_market_cap:.1f}십억원")
+    #
+    # with col4:
+    #     st.metric(label="종목 수", value=len(portfolio))
 
-def get_last_year_end():
-    """Get the last year's end date."""
-    today = datetime.now()
-    last_year = today.year - 1
-    last_year_end = datetime(last_year, 12, 31)
-    return last_year_end.date()