diff --git a/streamlit_quant/__init__.py b/streamlit_quant/__init__.py deleted file mode 100644 index e6c67f5..0000000 --- a/streamlit_quant/__init__.py +++ /dev/null @@ -1 +0,0 @@ -__all__ = ['backtest', 'strategy', 'quantcommon'] \ No newline at end of file diff --git a/streamlit_quant/app.py b/streamlit_quant/app.py deleted file mode 100644 index 2b3ef78..0000000 --- a/streamlit_quant/app.py +++ /dev/null @@ -1,13 +0,0 @@ -from src import streamlit as st - -crawling_page = st.Page("crawling.py", title="크롤링") -super_quality_page = st.Page("super_quality.py", title="슈퍼 퀄리티 전략") -super_value_momentum_page = st.Page("super_value_momentum.py", title="슈퍼 밸류 모멘텀 전략") - -pg = st.navigation({ - '크롤링': [crawling_page], - '전략': [super_quality_page, super_value_momentum_page], -}) - -st.set_page_config(page_title="콴트 매니저", page_icon=":material/edit:") -pg.run() \ No newline at end of file diff --git a/streamlit_quant/backtest/backtest-1.py b/streamlit_quant/backtest/backtest-1.py deleted file mode 100644 index 1f2731e..0000000 --- a/streamlit_quant/backtest/backtest-1.py +++ /dev/null @@ -1,48 +0,0 @@ -import bt -import matplotlib.pyplot as plt - -import pandas as pd - -from streamlit_quant import quantcommon -import streamlit_quant.strategy.multi_factor as multi_factor -import streamlit_quant.strategy.magic_formula as magic_formula - - -qc = quantcommon.QuantCommon() -mf = multi_factor.get_multi_factor_top(qc, 20) -magic_formula = magic_formula.get_magic_formula_top(20) - -codes = ','.join(magic_formula['종목코드'].array) -price = qc.get_price_list_by_code(codes) - -# price = price.set_index(['날짜']) -# price.rename(columns={"날짜": "Date"}) -price["Date"] = pd.to_datetime(price["날짜"]) - -pivot_df = price.pivot(index="Date", columns="종목코드", values="종가") - -# print(pivot_df.tail) - -strategy = bt.Strategy("Asset_EW", [ - bt.algos.SelectAll(), # 모든 데이터 사용 - bt.algos.WeighEqually(), # 동일 비중 투자 - bt.algos.RunMonthly(), # 매 월말 리밸런싱 - bt.algos.Rebalance() # 계산된 비중에 따라 리밸런싱 -]) - -# 가격 데이터 중 시작 시점이 모두 다르므로, dropna() 함수를 통해 NA를 모두 제거하여 시작 시점을 맞춤 -pivot_df.dropna(inplace=True) - -# 백테스트 생성 -backtest = bt.Backtest(strategy, pivot_df) - -# 백테스트 실행 -result = bt.run(backtest) -# prices: 누적 수익률이 데이터프레임 형태로 나타나며, 시작 시점을 100으로 환산하여 계산 -# to_returns: 수익률 계산 -# print(result.prices.to_returns()) - -result.plot(figsize=(10, 6), legend=False) -plt.show() - -result.display() \ No newline at end of file diff --git a/streamlit_quant/crawling.py b/streamlit_quant/crawling.py deleted file mode 100644 index 614233a..0000000 --- a/streamlit_quant/crawling.py +++ /dev/null @@ -1,6 +0,0 @@ -from src import streamlit as st - -st.button(label='동작1') -st.button(label='동작2') -st.button(label='동작3') -st.button(label='동작4') \ No newline at end of file diff --git a/streamlit_quant/quantcommon.py b/streamlit_quant/quantcommon.py deleted file mode 100644 index 27179c3..0000000 --- a/streamlit_quant/quantcommon.py +++ /dev/null @@ -1,140 +0,0 @@ -import os -from urllib.parse import quote_plus - -import pandas as pd -import pymysql -from dotenv import load_dotenv -from sqlalchemy import create_engine - - -class QuantCommon: - def __init__(self): - load_dotenv() - self.user = os.getenv('DB_USER') - self.pw = os.getenv('DB_PW') - self.engine_for_pw = quote_plus(self.pw) - self.host = os.getenv('DB_HOST') - self.port = int(os.getenv('DB_PORT')) - self.db = os.getenv('DB_DB') - - def create_engine(self): - return create_engine(f'mysql+pymysql://{self.user}:{self.engine_for_pw}@{self.host}:{self.port}/{self.db}') - - def connect(self): - return pymysql.connect(user=self.user, - passwd=self.pw, - host=self.host, - port=self.port, - db=self.db, - charset='utf8') - - def get_ticker_list(self): - engine = self.create_engine() - - try: - ticker_list = pd.read_sql(""" - select * from kor_ticker - where 기준일 = (select max(기준일) from kor_ticker) - and 종목구분 = '보통주'; - """, con=engine) - finally: - engine.dispose() - - return ticker_list - - def get_value_list(self): - engine = self.create_engine() - - try: - value_list = pd.read_sql(""" - select * from kor_value - where 기준일 = (select max(기준일) from kor_value); - """, con=engine) - finally: - engine.dispose() - - return value_list - - def get_price_list(self, interval_month): - engine = self.create_engine() - - try: - price_list = pd.read_sql(f""" - select 날짜, 종가, 종목코드 - from kor_price - where 날짜 >= (select (select max(날짜) from kor_price) - interval {interval_month} month); - """, con=engine) - finally: - engine.dispose() - - return price_list - - def get_price_list_by_code(self, codes): - engine = self.create_engine() - - try: - price_list = pd.read_sql(f""" - select * from kor_price - where 종목코드 in ({codes}); - """, con=engine) - finally: - engine.dispose() - - return price_list - - def get_fs_list(self): - engine = self.create_engine() - - try: - fs_list = pd.read_sql(""" - select * from kor_fs - where 계정 in ('당기순이익', '매출총이익', '영업활동으로인한현금흐름', '자산', '자본') - and 공시구분 = 'q'; - """, con=engine) - finally: - engine.dispose() - - return fs_list - - def get_fs_list_by_account_and_date(self, account, date): - engine = self.create_engine() - - try: - fs_list = pd.read_sql(f""" - select * from kor_fs - where 계정 in ({account}) - and 기준일 in ({date}) - and 공시구분 = 'y'; - """, con=engine) - finally: - engine.dispose() - - return fs_list - - def get_expanded_fs_list(self): - engine = self.create_engine() - - try: - fs_list = pd.read_sql(""" - select * from kor_fs - where 계정 in ('매출액', '당기순이익', '법인세비용', '이자비용', '현금및현금성자산', - '부채', '유동부채', '유동자산', '비유동자산', '감가상각비') - and 공시구분 = 'q'; - """, con=engine) - finally: - engine.dispose() - - 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/src/current-financial-statements.py b/streamlit_quant/src/current-financial-statements.py deleted file mode 100644 index ddaf748..0000000 --- a/streamlit_quant/src/current-financial-statements.py +++ /dev/null @@ -1,131 +0,0 @@ -import re -import time - -import pandas as pd -import requests as rq -from bs4 import BeautifulSoup -from tqdm import tqdm - -from streamlit_quant import quantcommon - - -# 재무제표 크롤링 - -def get_ticker_list(): - engine = quantcommon.QuantCommon().create_engine() - # 티커리스트 불러오기 - ticker_list = {} - try: - ticker_list = pd.read_sql(""" - select * from kor_ticker - where 기준일 = (select max(기준일) from kor_ticker) - and 종목구분 = '보통주'; - """, con=engine) - finally: - engine.dispose() - - return ticker_list - - -# 재무제표 클렌징 함수 -def clean_fs(df, ticker, frequency): - df = df[~df.loc[:, ~df.columns.isin(['계정'])].isna().all(axis=1)] - df = df.drop_duplicates(['계정'], keep='first') - df = pd.melt(df, id_vars='계정', var_name='기준일', value_name='값') - df = df[~pd.isnull(df['값'])] - df['계정'] = df['계정'].replace({'계산에 참여한 계정 펼치기': ''}, regex=True) - df['기준일'] = pd.to_datetime(df['기준일'], - format='%Y/%m') + pd.tseries.offsets.MonthEnd() - df['종목코드'] = ticker - df['공시구분'] = frequency - - return df - - -# ticker 별 재무제표 조회해서 DB에 저장 -def process_for_fs(ticker_list): - # DB 연결 - common = quantcommon.QuantCommon() - engine = common.create_engine() - con = common.connect() - mycursor = con.cursor() - - # DB 저장 쿼리 - query = """ - insert into kor_fs (계정, 기준일, 값, 종목코드, 공시구분) - values (%s,%s,%s,%s,%s) as new - on duplicate key update - 값=new.값 - """ - - # 오류 발생시 저장할 리스트 생성 - error_list = [] - - # for loop - for i in tqdm(range(0, len(ticker_list))): - - # 티커 선택 - ticker = ticker_list['종목코드'][i] - - # 오류 발생 시 이를 무시하고 다음 루프로 진행 - try: - # url 생성 - url = f'https://comp.fnguide.com/SVO2/ASP/SVD_Finance.asp?pGB=1&gicode=A{ticker}' - - # 데이터 받아오기 - data = pd.read_html(url, displayed_only=False) - - # 연간 데이터 - data_fs_y = pd.concat([ - data[0].iloc[:, ~data[0].columns.str.contains('전년동기')], data[2], - data[4] - ]) - data_fs_y = data_fs_y.rename(columns={data_fs_y.columns[0]: "계정"}) - - # 결산년 찾기 - page_data = rq.get(url) - page_data_html = BeautifulSoup(page_data.content, 'html.parser') - - fiscal_data = page_data_html.select('div.corp_group1 > h2') - fiscal_data_text = fiscal_data[1].text - fiscal_data_text = re.findall('[0-9]+', fiscal_data_text) - - # 결산년에 해당하는 계정만 남기기 - data_fs_y = data_fs_y.loc[:, (data_fs_y.columns == '계정') | ( - data_fs_y.columns.str[-2:].isin(fiscal_data_text))] - - # 클렌징 - data_fs_y_clean = clean_fs(data_fs_y, ticker, 'y') - - # 분기 데이터 - data_fs_q = pd.concat([ - data[1].iloc[:, ~data[1].columns.str.contains('전년동기')], data[3], - data[5] - ]) - data_fs_q = data_fs_q.rename(columns={data_fs_q.columns[0]: "계정"}) - - data_fs_q_clean = clean_fs(data_fs_q, ticker, 'q') - - # 두개 합치기 - data_fs_bind = pd.concat([data_fs_y_clean, data_fs_q_clean]) - - # 재무제표 데이터를 DB에 저장 - args = data_fs_bind.values.tolist() - mycursor.executemany(query, args) - con.commit() - - except: - # 오류 발생시 해당 종목명을 저장하고 다음 루프로 이동 - print(ticker) - error_list.append(ticker) - - # 타임슬립 적용 - time.sleep(2) - - # DB 연결 종료 - engine.dispose() - con.close() - -if __name__ == '__main__': - tickers = get_ticker_list() - process_for_fs(tickers) \ No newline at end of file diff --git a/streamlit_quant/src/current-price.py b/streamlit_quant/src/current-price.py deleted file mode 100644 index 34cb3e2..0000000 --- a/streamlit_quant/src/current-price.py +++ /dev/null @@ -1,101 +0,0 @@ -# 패키지 불러오기 - -import time -from datetime import date -from io import BytesIO - -import pandas as pd -import requests as rq -from tqdm import tqdm - -from streamlit_quant import quantcommon - - -# 주가 크롤링 - -def get_ticker_list(): - engine = quantcommon.QuantCommon().create_engine() - # 티커리스트 불러오기 - ticker_list = {} - try: - ticker_list = pd.read_sql(""" - select * from kor_ticker - where 기준일 = (select max(기준일) from kor_ticker) - and 종목구분 = '보통주'; - """, con=engine) - finally: - engine.dispose() - - return ticker_list - - -def process_for_price(ticker_list): - # DB 저장 쿼리 - query = """ - insert into kor_price (날짜, 시가, 고가, 저가, 종가, 거래량, 종목코드) - values (%s,%s,%s,%s,%s,%s,%s) as new - on duplicate key update - 시가 = new.시가, 고가 = new.고가, 저가 = new.저가, - 종가 = new.종가, 거래량 = new.거래량; - """ - - # DB 연결 - common = quantcommon.QuantCommon() - engine = common.create_engine() - con = common.connect() - - mycursor = con.cursor() - # 오류 발생시 저장할 리스트 생성 - error_list = [] - - # 전종목 주가 다운로드 및 저장 - for i in tqdm(range(0, len(ticker_list))): - - # 티커 선택 - ticker = ticker_list['종목코드'][i] - - # 시작일과 종료일 - # fr = (date.today() + relativedelta(years=-5)).strftime("%Y%m%d") - to = (date.today()).strftime("%Y%m%d") - fr = '20250125' - - # 오류 발생 시 이를 무시하고 다음 루프로 진행 - try: - - # url 생성 - url = f'''https://fchart.stock.naver.com/siseJson.nhn?symbol={ticker}&requestType=1 - &startTime={fr}&endTime={to}&timeframe=day''' - - # 데이터 다운로드 - data = rq.get(url).content - data_price = pd.read_csv(BytesIO(data)) - - # 데이터 클렌징 - price = data_price.iloc[:, 0:6] - price.columns = ['날짜', '시가', '고가', '저가', '종가', '거래량'] - price = price.dropna() - price['날짜'] = price['날짜'].str.extract("(\d+)") - price['날짜'] = pd.to_datetime(price['날짜']) - price['종목코드'] = ticker - - # 주가 데이터를 DB에 저장 - args = price.values.tolist() - mycursor.executemany(query, args) - con.commit() - - except: - - # 오류 발생시 error_list에 티커 저장하고 넘어가기 - print(ticker) - error_list.append(ticker) - - # 타임슬립 적용 - time.sleep(2) - - # DB 연결 종료 - engine.dispose() - con.close() - -if __name__ == '__main__': - ticker_list = get_ticker_list() - process_for_price(ticker_list) \ No newline at end of file diff --git a/streamlit_quant/src/current-stock.py b/streamlit_quant/src/current-stock.py deleted file mode 100644 index 46c1d2d..0000000 --- a/streamlit_quant/src/current-stock.py +++ /dev/null @@ -1,193 +0,0 @@ -import re -import time -from io import BytesIO - -import numpy as np -import pandas as pd -import requests as rq -from tqdm import tqdm -from bs4 import BeautifulSoup -from dotenv import load_dotenv -from streamlit_quant import quantcommon - -load_dotenv() - -GEN_OTP_URL = 'http://data.krx.co.kr/comm/fileDn/GenerateOTP/generate.cmd' -DOWN_URL = 'http://data.krx.co.kr/comm/fileDn/download_csv/download.cmd' - -# 최근 영업일을 가져옴 -def get_latest_biz_day(): - url = 'https://finance.naver.com/sise/sise_deposit.nhn' - data = rq.post(url) - data_html = BeautifulSoup(data.content, 'lxml') - parse_day = data_html.select_one('div.subtop_sise_graph2 > ul.subtop_chart_note > li > span.tah').text - biz_day = re.findall('[0-9]+', parse_day) - biz_day = ''.join(biz_day) - return biz_day - - -# 업종 분류 현황 가져옴 -def get_stock_data(biz_day, mkt_id): - # logging.basicConfig(level=logging.DEBUG) - gen_otp_data = { - 'locale': 'ko_KR', - 'mktId': mkt_id, # STK: 코스피, KSQ: 코스닥 - 'trdDd': biz_day, - 'money': '1', - 'csvxls_isNo': 'false', - 'name': 'fileDown', - 'url': 'dbms/MDC/STAT/standard/MDCSTAT03901' - } - headers = { - 'Referer': 'http://data.krx.co.kr/contents/MDC/MDI/mdiLoader/index.cmd?menuId=MDC0201050201', - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36' - } - - otp = rq.post(url=GEN_OTP_URL, data=gen_otp_data, headers=headers, verify=False) - # # 요청 디버깅 - # print("===== Request Details =====") - # print(f"Method: {otp.request.method}") - # print(f"URL: {otp.request.url}") - # print(f"Headers: {otp.request.headers}") - # print(f"Body: {otp.request.body}") - # - # # 응답 디버깅 - # print("===== Response Details =====") - # print(f"Status Code: {otp.status_code}") - # print(f"Headers: {otp.headers}") - # print(f"Body: {otp.text}") - - down_sector = rq.post(url=DOWN_URL, data={'code': otp.text}, headers=headers) - return pd.read_csv(BytesIO(down_sector.content), encoding='EUC-KR') - - -# 개별 지표 조회 -def get_ind_stock_data(biz_day): - gen_otp_data = { - 'locale': 'ko_KR', - 'searchType': '1', - 'mktId': 'ALL', - 'trdDd': biz_day, - 'csvxls_isNo': 'false', - 'name': 'fileDown', - 'url': 'dbms/MDC/STAT/standard/MDCSTAT03501' - } - headers = { - 'Referer': 'http://data.krx.co.kr/contents/MDC/MDI/mdiLoader/index.cmd?menuId=MDC0201050201', - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36' - } - - otp = rq.post(url=GEN_OTP_URL, data=gen_otp_data, headers=headers, verify=False) - - down_ind_sector = rq.post(url=DOWN_URL, data={'code': otp.text}, headers=headers) - return pd.read_csv(BytesIO(down_ind_sector.content), encoding='EUC-KR') - - -def process_for_total_stock(biz_day): - # 업종 분류 현황(코스피, 코스닥) - sector_stk = get_stock_data(biz_day, 'STK') - sector_ksq = get_stock_data(biz_day, 'KSQ') - # 각각 조회 후 합침 - krx_sector = pd.concat([sector_stk, sector_ksq]).reset_index(drop=True) - krx_sector['종목명'] = krx_sector['종목명'].str.strip() - krx_sector['기준일'] = biz_day - - # 개별 지표 조회 - krx_ind = get_ind_stock_data(biz_day) - krx_ind['종목명'] = krx_ind['종목명'].str.strip() - krx_ind['기준일'] = biz_day - - # 데이터 정리 - # 종목, 개별 중 한군데만 있는 데이터 삭제(선박펀드, 광물펀드, 해외종목 등) - diff = list(set(krx_sector['종목명']).symmetric_difference(set(krx_ind['종목명']))) - - kor_ticker = pd.merge(krx_sector, - krx_ind, - on=krx_sector.columns.intersection( - krx_ind.columns).tolist(), - how='outer') - # 일반적인 종목과 SPAC, 우선주, 리츠, 기타 주식을 구분 - kor_ticker['종목구분'] = np.where(kor_ticker['종목명'].str.contains('스팩|제[0-9]+호'), '스팩', - np.where(kor_ticker['종목코드'].str[-1:] != '0', '우선주', - np.where(kor_ticker['종목명'].str.endswith('리츠'), '리츠', - np.where(kor_ticker['종목명'].isin(diff), '기타', - '보통주')))) - kor_ticker = kor_ticker.reset_index(drop=True) - kor_ticker.columns = kor_ticker.columns.str.replace(' ', '') - kor_ticker = kor_ticker[['종목코드', '종목명', '시장구분', '종가', - '시가총액', '기준일', 'EPS', '선행EPS', 'BPS', '주당배당금', '종목구분']] - kor_ticker = kor_ticker.replace({np.nan: None}) - kor_ticker['기준일'] = pd.to_datetime(kor_ticker['기준일']) - - save_ticker(kor_ticker) - - -def save_ticker(ticker): - con = quantcommon.QuantCommon().connect() - - mycursor = con.cursor() - query = f""" - insert into kor_ticker (종목코드,종목명,시장구분,종가,시가총액,기준일,EPS,선행EPS,BPS,주당배당금,종목구분) - values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) as new - on duplicate key update - 종목명=new.종목명,시장구분=new.시장구분,종가=new.종가,시가총액=new.시가총액,EPS=new.EPS,선행EPS=new.선행EPS, - BPS=new.BPS,주당배당금=new.주당배당금,종목구분 = new.종목구분; - """ - - args = ticker.values.tolist() - - mycursor.executemany(query, args) - con.commit() - - con.close() - - -# WICS 기준 섹터정보 크롤링 -def process_for_wics(biz_day): - sector_code = [ - 'G25', 'G35', 'G50', 'G40', 'G10', 'G20', 'G55', 'G30', 'G15', 'G45' - ] - - data_sector = [] - - # 모든 섹터에 대한 데이터 받아서 가공 - for i in tqdm(sector_code): - url = f'''http://www.wiseindex.com/Index/GetIndexComponets?ceil_yn=0&dt={biz_day}&sec_cd={i}''' - data = rq.get(url).json() - data_pd = pd.json_normalize(data['list']) - - data_sector.append(data_pd) - - time.sleep(2) - - kor_sector = pd.concat(data_sector, axis=0) - kor_sector = kor_sector[['IDX_CD', 'CMP_CD', 'CMP_KOR', 'SEC_NM_KOR']] - kor_sector['기준일'] = biz_day - kor_sector['기준일'] = pd.to_datetime(kor_sector['기준일']) - save_sector(kor_sector) - - -def save_sector(sector): - con = quantcommon.QuantCommon().connect() - - mycursor = con.cursor() - query = f""" - insert into kor_sector (IDX_CD, CMP_CD, CMP_KOR, SEC_NM_KOR, 기준일) - values (%s,%s,%s,%s,%s) as new - on duplicate key update - IDX_CD = new.IDX_CD, CMP_KOR = new.CMP_KOR, SEC_NM_KOR = new.SEC_NM_KOR - """ - - args = sector.values.tolist() - - mycursor.executemany(query, args) - con.commit() - - con.close() - - -if __name__ == '__main__': - # sector와 ticker 갱신 - latest_biz_day = get_latest_biz_day() - process_for_total_stock(latest_biz_day) - process_for_wics(latest_biz_day) diff --git a/streamlit_quant/strategy/__init__.py b/streamlit_quant/strategy/__init__.py deleted file mode 100644 index 8a05387..0000000 --- a/streamlit_quant/strategy/__init__.py +++ /dev/null @@ -1 +0,0 @@ -__all__ = ['multi_factor', 'f_score'] \ No newline at end of file diff --git a/streamlit_quant/strategy/all_value.py b/streamlit_quant/strategy/all_value.py deleted file mode 100644 index 4b35a8b..0000000 --- a/streamlit_quant/strategy/all_value.py +++ /dev/null @@ -1,37 +0,0 @@ -import matplotlib.pyplot as plt -import numpy as np -import seaborn as sns -from streamlit_quant import quantcommon - - -#가치주 포트폴리오. PER, PBR, PCR, PSR, DY -def get_all_value_top(count): - qc = quantcommon.QuantCommon() - ticker_list = qc.get_ticker_list() - value_list = qc.get_value_list() - - # 가치 지표가 0이하인 경우 nan으로 변경 - value_list.loc[value_list['값'] <= 0, '값'] = np.nan - # 가치지표 테이블을 가로로 긴 형태로 변경 - value_pivot = value_list.pivot(index='종목코드', columns='지표', values='값') - # 티커 테이블과 가치 지표 테이블을 합침 - data_bind = ticker_list[['종목코드', '종목명']].merge(value_pivot, - how='left', - on='종목코드') - value_list_copy = data_bind.copy() - # DY(배당수익률)만 높을수록 좋은 지표라서 역수 - value_list_copy['DY'] = 1 / value_list_copy['DY'] - value_list_copy = value_list_copy[['PER', 'PBR', 'PCR', 'PSR', 'DY']] - value_rank_all = value_list_copy.rank(axis=0) - mask = np.triu(value_rank_all.corr()) - - fig, ax = plt.subplots(figsize=(10, 6)) - sns.heatmap(value_rank_all.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() - value_sum_all = value_rank_all.sum(axis=1, skipna=False).rank() - - return data_bind.loc[value_sum_all <= count] - -if __name__ == '__main__': - print(get_all_value_top(20)) \ No newline at end of file diff --git a/streamlit_quant/strategy/f_score.py b/streamlit_quant/strategy/f_score.py deleted file mode 100644 index e54d0d7..0000000 --- a/streamlit_quant/strategy/f_score.py +++ /dev/null @@ -1,104 +0,0 @@ -from datetime import datetime -import pandas as pd -from streamlit_quant import quantcommon - - -# 흑자 기업이면 1점(당기순이익) -def calc_net_income(qc, base_date): - net_income_list = qc.get_fs_list_by_account_and_date("'당기순이익'", f"'{base_date}'") - net_income_list['score1'] = (net_income_list['값'] > 0).astype(int) - - return net_income_list[['종목코드', 'score1']] - - -# CFO(영업활동현금흐름) 흑자 기업이면 1점 -def calc_cfo(qc, base_date): - cfo_list = qc.get_fs_list_by_account_and_date("'*영업에서창출된현금흐름'", f"'{base_date}'") - cfo_list['score2'] = (cfo_list['값'] > 0).astype(int) - - return cfo_list[['종목코드', 'score2']] - - -# 신규 주식 발행(유상증사): 전년 없음인 경우 1점 -# 제작년과 작년 자본금 변화가 없는 경우로 체크 -def calc_capital(qc, base_date): - last_year = datetime(base_date.year - 1, base_date.month, base_date.day).date() - capital_date = f"'{last_year}', '{base_date}'" - - # 자본금 - capital_list = qc.get_fs_list_by_account_and_date("'자본금'", capital_date) - - # 기준일별로 피벗 테이블 생성 - pivot_df = capital_list.pivot_table( - values='값', - index='종목코드', - columns='기준일', - aggfunc='first' - ) - pivot_df = pivot_df.dropna() - - # 값 차이 계산 및 score 부여 - pivot_df['diff'] = pivot_df[base_date] - pivot_df[last_year] - pivot_df['score3'] = (pivot_df['diff'] == 0).astype(int) - - # 결과 정리 - 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() - # 시가총액을 기준으로 정렬 - ticker_list['분류'] = pd.qcut(ticker_list['시가총액'], - q=[0, 0.2, 0.8, 1.0], # 0-20%, 20-80%, 80-100% 구간 - labels=['소형주', '중형주', '대형주']) - - return ticker_list[['종목코드', '종목명', '분류', '종가']] - -def get_f_score(qc, base_date): - ticker_list = get_ticker_list(qc) - 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') - merge_score1['score1'] = merge_score1['score1'].fillna(0).astype(int) - - # score 2 병합 + NaN인 경우 기본값 0 - merge_score2 = merge_score1.merge(score2_list, on='종목코드', how='left') - merge_score2['score2'] = merge_score2['score2'].fillna(0).astype(int) - - # score 3 병합 + NaN인 경우 기본값 0 - merge_score3 = merge_score2.merge(score3_list, on='종목코드', how='left') - merge_score3['score3'] = merge_score3['score3'].fillna(0).astype(int) - - # 개별 점수들로 신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 - -if __name__ == '__main__': - date = datetime(2024, 12, 31).date() - print(get_f_score(quantcommon.QuantCommon(), date).head(30)) \ No newline at end of file diff --git a/streamlit_quant/strategy/magic_formula.py b/streamlit_quant/strategy/magic_formula.py deleted file mode 100644 index d3f0760..0000000 --- a/streamlit_quant/strategy/magic_formula.py +++ /dev/null @@ -1,77 +0,0 @@ -import numpy as np -from streamlit_quant import quantcommon - - -# 마법 공식 포트폴리오. 밸류와 퀄리티의 조합. 조엘 그린블라트의 '마법공식' -def get_magic_formula_top(count): - qc = quantcommon.QuantCommon() - ticker_list = qc.get_ticker_list() - fs_list = qc.get_expanded_fs_list() - - fs_list = fs_list.sort_values(['종목코드', '계정', '기준일']) - # TTM 값을 구하기 위해서 rolling() 메소드를 통해 4분기 합 구함. 4분기 데이터가 없는 경우 제외하기 위해서 min_periods=4 - 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') - - data_bind = ticker_list[['종목코드', '종목명', '시가총액']].merge(fs_list_pivot, - how='left', - on='종목코드') - # 티커 테이블에 있는 시가총액은 원, 제무제표 테이블은 억 단위이므로 단위를 맞춤 - data_bind['시가총액'] = data_bind['시가총액'] / 100000000 - - # 이익수익률 계산식: 이자 및 법인세 차감전 이익(EBIT) / 가입가치(시가총액 + 순차입금) - # 분자(EBIT) - magic_ebit = data_bind['당기순이익'] + data_bind['법인세비용'] + data_bind['이자비용'] - - # 분모 - magic_cap = data_bind['시가총액'] - magic_debt = data_bind['부채'] - - ## 분모: 여유자금 - magic_excess_cash = data_bind['유동부채'] - data_bind['유동자산'] + data_bind[ - '현금및현금성자산'] - magic_excess_cash[magic_excess_cash < 0] = 0 - magic_excess_cash_final = data_bind['현금및현금성자산'] - magic_excess_cash - - magic_ev = magic_cap + magic_debt - magic_excess_cash_final - - # 이익수익률 - magic_ey = magic_ebit / magic_ev - - # 투하자본 수익률 - magic_ic = (data_bind['유동자산'] - data_bind['유동부채']) + (data_bind['비유동자산'] - - data_bind['감가상각비']) - magic_roc = magic_ebit / magic_ic - - # 열 입력하기 - data_bind['이익 수익률'] = magic_ey - data_bind['투하자본 수익률'] = magic_roc - - magic_rank = (magic_ey.rank(ascending=False, axis=0) + - magic_roc.rank(ascending=False, axis=0)).rank(axis=0) - return data_bind.loc[magic_rank <= 20, ['종목코드', '종목명', '이익 수익률', '투하자본 수익률']].round(4) - - # data_bind['투자구분'] = np.where(magic_rank <= 20, '마법공식', '기타') - # - # plt.rc('font', family='Malgun Gothic') - # plt.subplots(1, 1, figsize=(10, 6)) - # sns.scatterplot(data=data_bind, - # x='이익 수익률', - # y='투하자본 수익률', - # hue='투자구분', - # style='투자구분', - # s=200) - # plt.xlim(0, 1) - # plt.ylim(0, 1) - # plt.show() - -if __name__ == '__main__': - print(get_magic_formula_top(20)) \ No newline at end of file diff --git a/streamlit_quant/strategy/momentum.py b/streamlit_quant/strategy/momentum.py deleted file mode 100644 index 8f66225..0000000 --- a/streamlit_quant/strategy/momentum.py +++ /dev/null @@ -1,88 +0,0 @@ -import pandas as pd -import matplotlib.pyplot as plt -import seaborn as sns -import statsmodels.api as sm -import numpy as np -from streamlit_quant import quantcommon - - -def print_graph(values): - plt.rc('font', family='Malgun Gothic') - g = sns.relplot(data=values, - x='날짜', - y='종가', - col='종목코드', - col_wrap=5, - kind='line', - facet_kws={ - 'sharey': False, - 'sharex': True - }) - g.set(xticklabels=[]) - g.set(xlabel=None) - g.set(ylabel=None) - g.fig.set_figwidth(15) - g.fig.set_figheight(8) - plt.subplots_adjust(wspace=0.5, hspace=0.2) - plt.show() - -# strategy/momentum에 구현 -# 모멘텀 포트폴리오. 최근 12개월 수익률이 높은 주식 -def get_momentum_top(count): - qc = quantcommon.QuantCommon() - ticker_list = qc.get_ticker_list() - price_list = qc.get_price_list(interval_month=12) - - price_pivot = price_list.pivot(index='날짜', columns='종목코드', values='종가') - - # 가격 테이블에서 (가장 끝 행 / 가장 첫 행)으로 각 종목의 12개월 수익률을 구함 - ret_list = pd.DataFrame(data=(price_pivot.iloc[-1] / price_pivot.iloc[0]) - 1, - columns=['return']) - data_bind = ticker_list[['종목코드', '종목명']].merge(ret_list, how='left', on='종목코드') - - # 12개월 수익률 열 순위를 구함. 지표가 높을 수록 좋으니 ascending=False - momentum_rank = data_bind['return'].rank(axis=0, ascending=False) - # 모멘텀만 가지고 순위 측정 - price_momentum = price_list[price_list['종목코드'].isin( - data_bind.loc[momentum_rank <= count, '종목코드'])] - # 해당 종목들(모멘텀 상위 count 개)의 가격 그래프 확인 - # print_graph(price_momentum) - - # k-ratio(모멘텀의 꾸준함 지표) - # pct_change() 함수로 각 종목의 수익률 계산하고 수익률이 곗나되지 않는 첫 번째 행은 제외 - ret = price_pivot.pct_change().iloc[1:] - # 로그 누적 수익률 계산 - ret_cum = np.log(1 + ret).cumsum() - - # x축은 기간 - x = np.array(range(len(ret))) - k_ratio = {} - - for i in range(0, len(ticker_list)): - - ticker = data_bind.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() - - data_bind = data_bind.merge(k_ratio_bind, how='left', on='종목코드') - k_ratio_rank = data_bind['K_ratio'].rank(axis=0, ascending=False) - momentum_top = data_bind[k_ratio_rank <= count] - - k_ratio_momentum = price_list[price_list['종목코드'].isin(data_bind.loc[k_ratio_rank <= count, '종목코드'])] - print_graph(k_ratio_momentum) - return momentum_top - -if __name__ == '__main__': - print(get_momentum_top(20)) \ No newline at end of file diff --git a/streamlit_quant/strategy/multi_factor.py b/streamlit_quant/strategy/multi_factor.py deleted file mode 100644 index a0d9139..0000000 --- a/streamlit_quant/strategy/multi_factor.py +++ /dev/null @@ -1,251 +0,0 @@ -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 - -# 멀티 팩터 포트폴리오. -# 퀄리티: 자기자본이익률(ROE), 매출총이익(GPA), 영업활동현금흐름(CFO) -# 밸류: PER, PBR, PSR, PCR, DY -# 모멘텀: 12개월 수익률, K-Ratio -# 각 섹터별 아웃라이어를 제거한 후 순위와 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(qc, count): - 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() <= count, '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) \ No newline at end of file diff --git a/streamlit_quant/strategy/quality.py b/streamlit_quant/strategy/quality.py deleted file mode 100644 index 60e98a8..0000000 --- a/streamlit_quant/strategy/quality.py +++ /dev/null @@ -1,58 +0,0 @@ -import numpy as np -import matplotlib.pyplot as plt -import seaborn as sns -from streamlit_quant import quantcommon - - -# 퀄리티(우량주) 포트폴리오. 영업수익성이 높은 주식 -def get_quality_top(count): - qc = quantcommon.QuantCommon() - ticker_list = qc.get_ticker_list() - fs_list = qc.get_fs_list() - - fs_list = fs_list.sort_values(['종목코드', '계정', '기준일']) - # TTM 값을 구하기 위해서 rolling() 메소드를 통해 4분기 합 구함. 4분기 데이터가 없는 경우 제외하기 위해서 min_periods=4 - 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']) - # tail(1)을 통해 종목코드와 계정별 최근 데이터만 선택 - fs_list_clean = fs_list_clean.groupby(['종목코드', '계정']).tail(1) - - fs_list_pivot = fs_list_clean.pivot(index='종목코드', columns='계정', values='ttm') - # 수익성 지표에 해당하는 ROE, GPA, CFO 계산 - 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['자산'] - - # 티커 테이블과 합침 - quality_list = ticker_list[['종목코드', '종목명']].merge(fs_list_pivot, - how='left', - on='종목코드') - # quality_list.round(count).head() - - quality_list_copy = quality_list[['ROE', 'GPA', 'CFO']].copy() - quality_rank = quality_list_copy.rank(ascending=False, axis=0) - - mask = np.triu(quality_rank.corr()) - fig, ax = plt.subplots(figsize=(10, 6)) - sns.heatmap(quality_rank.corr(), - annot=True, - mask=mask, - annot_kws={"size": 16}, - vmin=0, - vmax=1, - center=0.5, - cmap='coolwarm', - square=True) - ax.invert_yaxis() - - # 위에서 구한 3개 지표들의 순위를 더한 후 다시 순위를 매김 - quality_sum = quality_rank.sum(axis=1, skipna=False).rank() - # 최종 순위가 낮은 종목 선택 - return quality_list.loc[quality_sum <= count, ['종목코드', '종목명', 'ROE', 'GPA', 'CFO']].round(4) - -if __name__ == '__main__': - print(get_quality_top(20)) \ No newline at end of file diff --git a/streamlit_quant/strategy/value.py b/streamlit_quant/strategy/value.py deleted file mode 100644 index 24a8e8c..0000000 --- a/streamlit_quant/strategy/value.py +++ /dev/null @@ -1,27 +0,0 @@ -import numpy as np - -from streamlit_quant import quantcommon - - -#가치주 포트폴리오. PER, PBR이 낮은 회사 20개 -def get_value_top(count): - qc = quantcommon.QuantCommon() - ticker_list = qc.get_ticker_list() - value_list = qc.get_value_list() - - # 가치 지표가 0이하인 경우 nan으로 변경 - value_list.loc[value_list['값'] <= 0, '값'] = np.nan - # 가치지표 테이블을 가로로 긴 형태로 변경 - value_pivot = value_list.pivot(index='종목코드', columns='지표', values='값') - # 티커 테이블과 가치 지표 테이블을 합침 - data_bind = ticker_list[['종목코드', '종목명']].merge(value_pivot, - how='left', - on='종목코드') - # rank() 함수로 PER, PBR 열의 순위를 구함. axis=0을 입력하여 순위는 열 방향으로 구함.(PER, PBR 각각 순위) - value_rank = data_bind[['PER', 'PBR']].rank(axis = 0) - # axis=1을 통해서 위에서 구한 순위 랭크를 합침. 합친 것을 다시 rank() - value_sum = value_rank.sum(axis = 1, skipna = False).rank() - return data_bind.loc[value_sum <= count, ['종목코드', '종목명', 'PER', 'PBR']] - -if __name__ == '__main__': - print(get_value_top(20)) \ No newline at end of file diff --git a/streamlit_quant/super_quality.py b/streamlit_quant/super_quality.py deleted file mode 100644 index 84ada85..0000000 --- a/streamlit_quant/super_quality.py +++ /dev/null @@ -1,67 +0,0 @@ -from datetime import datetime -from src import streamlit as st - -from strategy import f_score -import quantcommon - - -# st.write(""" -# '신F-스코어 3점 + 고GP/A 전략'을 '강환국 슈퍼 퀄리티 전략'이라 명명한다. -# 이 전략은 신F-스코어가 3점인 종목을 매수하되, GP/A로 순위를 매겨서 순위가 높은 종목만 매수하는 것이다. -# 이 경우 한국에서 수익이 어땠을지 분석해보자. -# 연도별로 신F-스코어 3점을 충족하는 종목은 600-700개였다.\n -# 신F-스코어 3점 기업 내에서도 GP/A가 높은 종목이 3점 종목 평균보다 CAGR 기준으로 3-4% 더 높았다. -# 반대로 GP/A가 낮은 종목의 수익률은 상대적으로 저조했다.\n\n -# --- -# 투자 전략: 강환국 슈퍼 퀄리티 전략 1.0\n -# 레벨: 초, 중급\n -# 스타일: 퀄리티\n -# 기대 CAGR: 약 20%\n -# 매수 전략: -# - 신F-스코어 3점 종목만 매수\n -# - 여기에 GP/A 순위를 부여, 순위 높은 20-30종목을 매수\n -# 매도 전략: 연 1회 리밸런싱\n\n\n -# --- -# 지금까지 소개한 거의 모든 전략에서 소형주 전략이 전체 주식 수익률보다 높았다. -# 시가총액 하위 20% 종목의 CAGR을 분석해보았다.\n\n -# --- -# 투자 전략: 강환국 슈퍼 퀄리티 전략 2.0\n -# 레벨: 초, 중급\n -# 스타일: 퀄리티\n -# 기대 CAGR: 20% 이상\n -# 매수 전략: -# 아래 조건을 만족하는 20-30종목 매수\n -# - 신F-스코어 3점 종목만 매수\n -# - 여기에 GP/A 순위를 부여, 순위 높은 종목만 매수\n -# - 단, 소형주(시가총액 최저 20%)만 매수\n -# 매도 전략: 연 1회 리밸런싱\n -# --- -# 소형주 중 신F-스코어가 3점인 종목을 찾아보니 2004-2016년 구간에 80-100개 종목이 남았다. -# 그 주식들을 통째로 매수해도 CAGR 34.55%를 벌수 있었다! -# 정말 상당한 수익이다. -# 이 종목들을 다 샀으면 총 1,159개 종목 중 14개가 파산했다.(1.2%) -# 또 1년간 마이너스 수익을 기록한 종목이 29.7%였다.\n -# 신F-스코어가 3점인 종목 중 GP/A가 높은 종목 위주로 매수했으면 (1) CAGR도 조금 개선되고 (2) 최상 30개 종목을 매수했을 경우 선택받은 종목 360개 중 파산한 기업은 단 1개였다. -# F-스코어와 GP/A는 엄청난 잠재력을 지닌 콤비네이션임이 분명하다. -# """) - -def get_last_year_end(): - # 현재 날짜 가져오기 (2025년 3월 16일 기준) - today = datetime.now() - - # 작년 연도 계산 - last_year = today.year - 1 - - # 작년 12월 31일 생성 - last_year_end = datetime(last_year, 12, 31) - - return last_year_end.date() - -st.write("투자 전략: 강환국 슈퍼 퀄리티 전략 2.0") - -date = get_last_year_end() -data = f_score.get_f_score(quantcommon.QuantCommon(), date) - -config = {} - -st.dataframe(data, column_config=config, use_container_width=True) diff --git a/streamlit_quant/super_value_momentum.py b/streamlit_quant/super_value_momentum.py deleted file mode 100644 index e75c644..0000000 --- a/streamlit_quant/super_value_momentum.py +++ /dev/null @@ -1,3 +0,0 @@ -from src import streamlit as st - -st.write("슈퍼 밸류 모멘텀 전략 2.0") \ No newline at end of file