다음으로는 extract.py 에 대해서 버그 수정 및 몇가지 팩터데이터 추가, 코드의 리팩토링을 진행해봅니다.
Bug fix
__find_financial_indicator()
일단 2분기 사업보고서가 발행될때 즈음해서 조회를 하려고 하니 제대로 조회가 되지 않는다거나, 똑같은 연도를 가진 데이터가 2~3개씩 중복으로 들어가는 것을 확인했습니다.
이것은 재무데이터를 조회할 때, 제무데이터가 없음에도 불구하고 조회를 하려고해서 이슈가 발생했습니다.
원래는 extract.py 파일의 __find_financial_indicator() 메소드 내부에 아래의 코드가 있었는데, 여기서의 pass 를 continue 로 수정합니다.
만들면서 일단 pass로 적어도 놓고 코드를 완성시켰었는데, 이 때는 이슈가 없어서 확인하지 못했던 부분이었습니다.
if report is None: # 리포트가 없다면 반복문 종료하기
continue
else:
# 조건들에 대해 추출하기
condition1 = CONDITION.get_condition1(report)
condition2 = CONDITION.get_condition2(report)
condition3 = CONDITION.get_condition3(report)
condition4 = CONDITION.get_condition4(report)
condition5 = CONDITION.get_condition5(report)
condition6 = CONDITION.get_condition6(report)
condition7 = CONDITION.get_condition7(report)
condition8 = CONDITION.get_condition8(report)
# condition9 = CONDITION.get_condition9(report)
condition10 = CONDITION.get_condition10(report)
condition15 = CONDITION.get_condition15(report)그리고 이 조건문 이전 단계에, 올해에 해당하는 각 분기별 재무제표를 조회할 때, 해당 재무제표가 아직 나올시기가 아니라면, 바로 조회를 종료하는 로직도 추가하였습니다.
예를 들어, 2분기 재무제표가 나올 시기가 아니라면, 3분기나 4분기 재무제표도 존재할리가 없기에 바로 종료하게 합니다.
# 각 분기별 재무제표를 불러올때, 아직 안나왔다면 그냥 재무제표계산을 중단하기.
if year == today.year:
if report_name == "11012" and today.month <= 7: # 2분기
break
if report_name == "11014" and today.month <= 11: # 3분기
break
if report_name == "11011" and today.month <= 12: # 4분기
break
extract.py 내부의 __find_financial_indicator() 의 초반 로직은 이러한 형태입니다.
for j, report_name in enumerate(self.report_code):
time.sleep(0.1)
# 연결 재무제표 불러오기
try:
report = self.dart.finstate_all(stock_code, year, report_name, fs_div='CFS')
except SSLError: # 재시도
report = self.dart.finstate_all(stock_code, year, report_name, fs_div='CFS')
today = datetime.today().date()
# 각 분기별 재무제표를 불러올때, 아직 안나왔다면 그냥 재무제표계산을 중단하기.
if year == today.year:
if report_name == "11012" and today.month <= 7: # 2분기
break
if report_name == "11014" and today.month <= 11: # 3분기
break
if report_name == "11011" and today.month <= 12: # 4분기
break
if report is None: # 리포트가 없다면 반복문 종료하기
continue
else:
# 조건들에 대해 추출하기
condition1 = CONDITION.get_condition1(report)
condition2 = CONDITION.get_condition2(report)
condition3 = CONDITION.get_condition3(report)
condition4 = CONDITION.get_condition4(report)
condition5 = CONDITION.get_condition5(report)
condition6 = CONDITION.get_condition6(report)
condition7 = CONDITION.get_condition7(report)
condition8 = CONDITION.get_condition8(report)
condition10 = CONDITION.get_condition10(report)
condition15 = CONDITION.get_condition15(report)
.
.
.재무제표 조회 로직 수정
가끔가다가 제무제표 중에 매출총이익 과 자본총계 를 구할 수 없는 패턴이 있어서, 이를 보완할 로직을 추가해주었습니다.
__find_financial_indicator()
# 매출총이익 계산
if stock_code == '011810': # 매출총이익 항목이 없는 회사도 있다. 이 경우, 매출액 - 매출원가로 계산.
grossProfit[j] = revenue[j] - self.__check_index_error(report,
CONDITION.get_condition11(report))
elif stock_code == '008770':
grossProfit[j] = revenue[j] - self.__check_index_error(report,
CONDITION.get_condition14(report))
elif stock_code in no_report_list: # 매출총이익도 없고 이를 계산할 매출원가도 없다.
grossProfit[j] = -1
else:
grossProfit[j] = self.__check_index_error(report, condition5)
# 매출총이익이 에러핸들링으로 인해서 -1인 경우.
# 특히 IT 서비스 몇몇 기업의 경우 매출총이익항목이 별도로 없고 재료비 항목도 없다.
# 이 경우에는 그냥 매출액을 매출총이익으로 넣어준다.
# 예시로 네이버증권에서 NAVER 항목에 대해서 참조.
if grossProfit[j] == -1:
grossProfit[j] = revenue[j] # 계정에서 자본과부채의 총계 등으로 표현되는 경우, 순수 자본총계를 구할 수가 없어서 에러핸들링이 됨.
# 이때는 자산총계에서 부채총계를 뺴는 것으로 자본총계를 구해줄 수 있음.
if equity[j] == -1:
equity[j] = total_assets[j] - liabilities[j]팩터데이터 추가
다음은 몇가지 팩터데이터를 추가합니다. 그간 데이터를 쭉 보면서, 공부도 같이 병행했는데, 역시나 필요하다 싶은 데이터가 있어서 추가하게 되었습니다.
아래부터는 좀 이것저것 자잘하게, 많은 부분을 손봅니다. git을 사용하시는 분들은 브랜치를 새로 따서 진행하시는게 안전하게 할 수 있을 것 같습니다.
1. YoY, Year on Year, 연간 증가율
이전까지 구하던 증가율은 QoQ, 전분기 대비 증가율이었는데, 공부하다보니 전년 동분기 대비 증가율을 구해야할 필요를 느꼈습니다. 왜냐하면, 기업중에서는 계절성 사업으로 인하여 특정 분기(계절)이 되어야 매출이 올라가는 기업도 있었기 때문입니다.
기존의 증가율을 QoQ 증가율로 명시하고, 새롭게 YoY 증가율을 구해보겠습니다.
구하는 방법은 여러가지가 있었는데, 다만 QoQ도 그렇고, 이 증가율 계산이라는 것이 비교대상이 음수의 값이면 계산결과가 예상한 결과와 좀 달라지는 이슈가 있어서 여러가지 고민을 하였습니다.
마침 검색을 하니 스택오버플로우에 비슷한 고민을 한 사람들이 많이 있었는데요, 그중에 하나가 쓸만해서 가져와보았습니다.
https://stackoverflow.com/questions/66795087/handling-negative-values-in-pct-changes
1-1 그룹핑을 위해 분기 컬럼 추가하기
일단 계산을 편하게 하기 위해서, 분기 컬럼을 추가해야할 필요가 있었습니다.
모든 계산은, extract.py 에서 이루어집니다.
init()
class Extract:
def __init__(self):
self.factor_data = KoreanMarketFactorData()
self.dart = OpenDartReader(OPEN_DART_KEY) # config/api_key.py에서 api key의 설정이 필요함.
self.report_code = [
'11013', # "1분기보고서":
'11012', # "반기보고서":
'11014', # "3분기보고서":
'11011' # "사업보고서"
]
self.indicators = [
'유동자산',
'부채총계',
'자본총계',
'자산총계',
'매출액',
'매출총이익',
'영업이익',
'당기순이익',
'영업활동현금흐름',
'잉여현금흐름'
]
self.financial_column_header = ["종목코드", "연도", "시가총액", "분기"] + self.indicators
.
.
.생성자 내부에 어트리뷰트들 중에 self.financial_column_header 의 값중에 분기 를 추가합니다. 위치는 어느곳이나 상관없으나, 이 위치를 확인하시고, 분기데이터를 해당 위치에 넣어주어야 합니다.
1-2 재무제표 조회 시에 분기 구분 추가해주기
__find_financial_indicator()
빈껍데기 변수를 미리 선언해줍니다.
def __find_financial_indicator(self, stock_code, year):
current_assets = [0, 0, 0, 0] # 유동자산
liabilities = [0, 0, 0, 0] # 부채총계
equity = [0, 0, 0, 0] # 자본총계
total_assets = [0, 0, 0, 0] # 자산총계
revenue = [0, 0, 0, 0] # 매출액
grossProfit = [0, 0, 0, 0] # 매출총이익
income = [0, 0, 0, 0] # 영업이익
net_income = [0, 0, 0, 0] # 당기순이익
cfo = [0, 0, 0, 0] # 영업활동현금흐름
capex = [0, 0, 0, 0] # 유형자산의 증가
fcf = [0, 0, 0, 0] # 잉여현금흐름 : 영업활동의흐름 - 유형자산의 증가
market_cap = [0, 0, 0, 0] # 시가 총액
date_year = str(year) # 년도 변수 지정
quarter = 0 # 분기마지막 줄에 quarter 를 추가해주었습니다.
재무제표를 추출하는 로직 내부에서, 각 보고서별로 quarter를 1부터 4로 넣어줍니다.
if report_name == '11013': # 1분기
date_month = '03'
date_day = 31 # 일만 계산할꺼니까 이것만 숫자로 지정
quarter = 1
elif report_name == '11012': # 2분기
date_month = '06'
date_day = 30
cfo[j] = cfo[j] - cfo[j - 1] # 현금흐름은 2분기부터 시작
capex[j] = capex[j] - capex[j - 1]
quarter = 2
elif report_name == '11014': # 3분기
date_month = '09'
date_day = 30
cfo[j] = cfo[j] - (cfo[j - 1] + cfo[j - 2])
capex[j] = capex[j] - (capex[j - 1] + capex[j - 2])
quarter = 3
else: # 4분기. 1 ~ 3분기 데이터를 더한다음 사업보고서에서 빼야 함
date_month = '12'
date_day = 30
revenue[j] = revenue[j] - (revenue[0] + revenue[1] + revenue[2])
grossProfit[j] = grossProfit[j] - (grossProfit[0] + grossProfit[1] + grossProfit[2])
income[j] = income[j] - (income[0] + income[1] + income[2])
net_income[j] = net_income[j] - (net_income[0] + net_income[1] + net_income[2])
cfo[j] = cfo[j] - (cfo[j - 1] + cfo[j - 2] + cfo[j - 3])
capex[j] = capex[j] - (capex[j - 1] + capex[j - 2] + capex[j - 3])
quarter = 4이 메소드 하단에 record 변수에 넣어줄때, 아까 선언했던 컬럼위치에 맞게 분기값을 넣어줍니다.
# 각 데이터들을 순서대로 리스트를 만들어서 record변수에 저장한다.
record = [stock_code, date_string, market_cap[j], quarter, current_assets[j], liabilities[j], equity[j],
total_assets[j],
revenue[j], grossProfit[j], income[j], net_income[j], cfo[j],
fcf[j]]1-3 YoY, QoQ 증가율 계산하기
기존의 증가율을 QoQ 증가율이라고 알 수 있도록, 코드리팩토링까지 같이 진행합니다.
__calculate_indicator()
three_indicators = ['매출액', '영업이익', '당기순이익']
three_indicators_status = ['매출액 상태', '영업이익 상태', '당기순이익 상태']
three_qoq_growth_indicators = ['QoQ 매출액 증가율', 'QoQ 영업이익 증가율', 'QoQ 당기순이익 증가율']
three_yoy_growth_indicators = ['YoY 매출액 증가율', 'YoY 영업이익 증가율', 'YoY 당기순이익 증가율']기존의 indicator 리스트를 선언해서 사용하던 부분에, 위와 같이 명시합니다.
이번에 구해볼 증가율들은 매출액, 영업이익, 당기순이익 이며 필요 시에 똑같은 방법으로 매출총이익의 증가율도 구해볼 수 있을 것입니다. 일단 필요한건 위 세 개인 것 같아서 이것만 먼저...
다음으로 증가율을 계산합니다. 여기서 YoY계산 시에 아까 선언했던 분기 컬럼값으로 그룹핑을 하게 됩니다.
# 분기별 매출액 / 영업이익 / 당기순이익 증가율
for i in range(0, len(three_indicators)):
s = df_finance[three_indicators[i]].shift()
df_finance[three_qoq_growth_indicators[i]] = df_finance[three_indicators[i]].sub(s).div(s.abs()) * 100
# 전년동기대비 매출액 / 영업이익 / 당기순이익 증가율
for i in range(0, len(three_indicators)):
s = df_finance.groupby('분기')[three_indicators[i]].shift()
df_finance[three_yoy_growth_indicators[i]] = df_finance[three_indicators[i]].sub(s).div(s.abs()) * 100위에 링크 걸어드린 스택오버플로우의 내용을 보시면 계산내용에 대해 금방 알 수 있습니다. 이전값을 구해와서, 현재값에서 이전값을 빼고, 이전값의 절대값으로 나눈 후 100을 곱해주는 로직입니다.
QoQ는 이전값을 그냥 바로 구해주면되고, YoY는 groupby()를 이용하여 그룹핑을 해줍니다. 그룹핑하면 각 분기별로 그룹핑이 되어, 이전값은 곧 전년의 동기가 되어있습니다.
그룹핑을 해준 후에는 그냥 계산해주면 됩니다. 그룹핑되어진 상태라서 그런지 값이 제대로 들어가긴했습니다. 검증에 대해서는 이후에 좀 더 알아보도록 합니다.
이후 분기 컬럼을 삭제합니다.
# 분기열 삭제
df_finance.drop(['분기'], axis=1)이후 리턴값에 다시 명시를 해줍니다.
### 데이터들의 헤더 순서를 다시 지정해준다.
return df_temp.reindex(
columns=[
'종목코드', '연도', '시가총액',
'분기 PBR', 'PSR', 'PGPR', 'POR', '분기 PER', 'PCR', 'PFCR',
'분기 ROE', 'GP/A', 'NCAV/MC'
]
+ self.indicators +
[
'부채비율', '매출총이익률', three_qoq_growth_indicators[0], three_yoy_growth_indicators[0],
three_indicators_status[0],
'영업이익률', three_qoq_growth_indicators[1], three_yoy_growth_indicators[1],
three_indicators_status[1],
'당기순이익률', three_qoq_growth_indicators[2], three_yoy_growth_indicators[2],
three_indicators_status[2]
]
)컬럼순서는 편하신대로 기재하시면 됩니다.
2. PEG, Price Earning to Growth, 주가수익성장비율
다음으로는 PEG 를 구해봅니다.
PEG는 피터린치가 참고했다고 하는 데이터로, PER을 EPS 증가율로 나누고 100을 곱한 값입니다. 요컨대 PER가 높으면, 주당 순이익의 증가율이 그만큼 받쳐주어야지 기업이 제대로 성장하고 있다는 것을 알 수 있다 라는 것입니다.
예를들어 PER가 10이라면, 10년간 벌어야지 투자한금액이 본전이라는 뜻이죠. 그러면 순이익이 연간 10%의 폭으로 성장해나가야지 이것이 1로 맞아떨어지는 것입니다.
피터린치는 0.5정도면 무조건 사야하고, 1정도가 투자하기 적절한 성장주라고 합니다.
성장주를 알아보기 위해서 저는 이 데이터를 구현하기로 했습니다.
이럴러면 각 보고서별로 EPS를 구해야하는데, EPS는 당기순이익을 발행주식수로 나누어야합니다. 문제는 지금 제가 구현한 내용상으로는 각 보고서발행 시기별 발행주식수를 알기 어렵다는 것입니다.
그렇지만 각 분기별 당기순이익의 증가율은 알수있으니, PER를 당기순이익의 증가율로 나누어보면 얼추 비슷하게 나오지않을까 생각하여, 당기순이익 증가율 을 사용하고자 합니다.
(주당 순이익이란, 당기순이익을 발행주식수로 나눈 값이므로, 당기순이익의 증가율은 결국 주당순이익의 증가율과 동일할 것이라는 전제)
단, 공부를 하다보니 이 당기순이익에 조금 의심이 가는데, 왜냐하면 회사가 가진 어떠한 자산을 처분하였을때, 당기순이익으로 잡힌다는 것입니다. (기계를 처분한다던지, 토지나 건물을 처분한다던지...) 이렇게 회사의 판매활동으로 발생하는 이익 외의 수익도 잡히다보니 당기순이익 증가율보다는 영업이익 증가율 을 사용하는게 낫지않을까...? 라고 생각합니다만, 일단 피터린치가 참고했다하니 그냥 써보도록 하겠습니다.
2-1 PEG 구해보기
__calculate_indicator()
일단 빈 컬럼을 선언합니다.
# 분기별 PER
df['분기 PER'] = np.nan
# 분기별 PBR
df['분기 PBR'] = np.nan
df['분기 PEG'] = np.nan
df['PGPR'] = np.nan
df['PSR'] = np.nan
df['GP/A'] = np.nan
df['POR'] = np.nan
df['PCR'] = np.nan
df['PFCR'] = np.nan
df['NCAV/MC'] = np.nan
df['분기 ROE'] = np.nan분기 PEG 라고 컬럼명을 기재해주었습니다. 저는 일단 분기별 데이터로 구해볼까하는데, 어차피 저는 연간 증가율도 계산해놓았으니, 언제든지 제 편한대로 참조할 값을 바꿀 수 있습니다.
다음으로는 PEG 를 계산합니다. PEG는 저는 분기별 PER에 전분기 대비 당기순이익 증가율을 이용해서 구해보았습니다.
이때 주의점은, 이 코드는 증가율을 계산하는 코드 다음에 위치해야합니다. (증가율이 이미 계산되어있어야 계산할 수 있기 때문)
# PEG : PER / 당기순이익 증가율
df_finance["분기 PEG"] = df_finance['분기 PER'] / df_finance[three_qoq_growth_indicators[2]]당시 리턴값에 분기별 PEG 값의 위치를 지정해서 명시합니다.
### 데이터들의 헤더 순서를 다시 지정해준다.
return df_temp.reindex(
columns=[
'종목코드', '연도', '시가총액',
'분기 PBR', 'PSR', 'PGPR', 'POR', '분기 PER', '분기 PEG', 'PCR', 'PFCR',
'분기 ROE', 'GP/A', 'NCAV/MC'
]
+ self.indicators +
[
'부채비율', '매출총이익률', three_qoq_growth_indicators[0], three_yoy_growth_indicators[0],
three_indicators_status[0],
'영업이익률', three_qoq_growth_indicators[1], three_yoy_growth_indicators[1],
three_indicators_status[1],
'당기순이익률', three_qoq_growth_indicators[2], three_yoy_growth_indicators[2],
three_indicators_status[2]
]
)위와 같이 추가를 해보았습니다. 각 메소드들의 내용은 아래와 같습니다.
extract.py - __find_financial_indicator()
def __find_financial_indicator(self, stock_code, year):
current_assets = [0, 0, 0, 0] # 유동자산
liabilities = [0, 0, 0, 0] # 부채총계
equity = [0, 0, 0, 0] # 자본총계
total_assets = [0, 0, 0, 0] # 자산총계
revenue = [0, 0, 0, 0] # 매출액
grossProfit = [0, 0, 0, 0] # 매출총이익
income = [0, 0, 0, 0] # 영업이익
net_income = [0, 0, 0, 0] # 당기순이익
cfo = [0, 0, 0, 0] # 영업활동현금흐름
capex = [0, 0, 0, 0] # 유형자산의 증가
fcf = [0, 0, 0, 0] # 잉여현금흐름 : 영업활동의흐름 - 유형자산의 증가
market_cap = [0, 0, 0, 0] # 시가 총액
date_year = str(year) # 년도 변수 지정
quarter = 0 # 분기
no_report_list = ['035420', '035720', '036570', '017670', '251270', '263750', '030200', '293490',
'112040', '259960', '032640', '180640', '058850'] # 매출총이익 계산 못하는 회사들
data = []
for j, report_name in enumerate(self.report_code):
time.sleep(0.1)
# 연결 재무제표 불러오기
try:
report = self.dart.finstate_all(stock_code, year, report_name, fs_div='CFS')
except SSLError: # 재시도
report = self.dart.finstate_all(stock_code, year, report_name, fs_div='CFS')
today = datetime.today().date()
# 각 분기별 재무제표를 불러올때, 아직 안나왔다면 그냥 재무제표계산을 중단하기.
if year == today.year:
if report_name == "11012" and today.month <= 7: # 2분기
break
if report_name == "11014" and today.month <= 11: # 3분기
break
if report_name == "11011" and today.month <= 12: # 4분기
break
if report is None: # 리포트가 없다면 반복문 종료하기
continue
else:
# 조건들에 대해 추출하기
condition1 = CONDITION.get_condition1(report)
condition2 = CONDITION.get_condition2(report)
condition3 = CONDITION.get_condition3(report)
condition4 = CONDITION.get_condition4(report)
condition5 = CONDITION.get_condition5(report)
condition6 = CONDITION.get_condition6(report)
condition7 = CONDITION.get_condition7(report)
condition8 = CONDITION.get_condition8(report)
condition10 = CONDITION.get_condition10(report)
condition15 = CONDITION.get_condition15(report)
# 유동자산
current_assets[j] = self.__check_index_error(report, condition1)
# 부채총계
liabilities[j] = self.__check_index_error(report, condition2)
# 자본총계
equity[j] = self.__check_index_error(report, condition3)
# 매출액 계산
if stock_code == '003550': # LG의 경우, 매출이 쪼개져있으므로 매출원가 + 매출총이익을 더한다.
revenue[j] = self.__check_index_error(report, CONDITION.get_condition11(report)) + \
self.__check_index_error(report, condition5)
else:
revenue[j] = self.__check_index_error(report, condition4)
# 매출총이익 계산
if stock_code == '011810': # 매출총이익 항목이 없는 회사도 있다. 이 경우, 매출액 - 매출원가로 계산.
grossProfit[j] = revenue[j] - self.__check_index_error(report,
CONDITION.get_condition11(report))
elif stock_code == '008770':
grossProfit[j] = revenue[j] - self.__check_index_error(report,
CONDITION.get_condition14(report))
elif stock_code in no_report_list: # 매출총이익도 없고 이를 계산할 매출원가도 없다.
grossProfit[j] = -1
else:
grossProfit[j] = self.__check_index_error(report, condition5)
# 매출총이익이 에러핸들링으로 인해서 -1인 경우.
# 특히 IT 서비스 몇몇 기업의 경우 매출총이익항목이 별도로 없고 재료비 항목도 없다.
# 이 경우에는 그냥 매출액을 매출총이익으로 넣어준다.
# 예시로 네이버증권에서 NAVER 항목에 대해서 참조.
if grossProfit[j] == -1:
grossProfit[j] = revenue[j]
# 영업이익
income[j] = self.__check_index_error(report, condition6)
# 당기순이익
if stock_code == '008600': # 법인세 차감전 금액에서 법인세 비용을 차감
net_income[j] = self.__check_index_error(report, CONDITION.get_condition12(report))\
- self.__check_index_error(report, CONDITION.get_condition13(report))
else:
net_income[j] = self.__check_index_error(report, condition7)
# 영업활동 현금흐름
cfo[j] = self.__check_index_error(report, condition8)
# 유형자산의 증가
capex[j] = self.__check_index_error(report, condition15)
# 자산총계
total_assets[j] = self.__check_index_error(report, condition10)
# 계정에서 자본과부채의 총계 등으로 표현되는 경우, 순수 자본총계를 구할 수가 없어서 에러핸들링이 됨.
# 이때는 자산총계에서 부채총계를 뺴는 것으로 자본총계를 구해줄 수 있음.
if equity[j] == -1:
equity[j] = total_assets[j] - liabilities[j]
if report_name == '11013': # 1분기
date_month = '03'
date_day = 31 # 일만 계산할꺼니까 이것만 숫자로 지정
quarter = 1
elif report_name == '11012': # 2분기
date_month = '06'
date_day = 30
cfo[j] = cfo[j] - cfo[j - 1] # 현금흐름은 2분기부터 시작
capex[j] = capex[j] - capex[j - 1]
quarter = 2
elif report_name == '11014': # 3분기
date_month = '09'
date_day = 30
cfo[j] = cfo[j] - (cfo[j - 1] + cfo[j - 2])
capex[j] = capex[j] - (capex[j - 1] + capex[j - 2])
quarter = 3
else: # 4분기. 1 ~ 3분기 데이터를 더한다음 사업보고서에서 빼야 함
date_month = '12'
date_day = 30
revenue[j] = revenue[j] - (revenue[0] + revenue[1] + revenue[2])
grossProfit[j] = grossProfit[j] - (grossProfit[0] + grossProfit[1] + grossProfit[2])
income[j] = income[j] - (income[0] + income[1] + income[2])
net_income[j] = net_income[j] - (net_income[0] + net_income[1] + net_income[2])
cfo[j] = cfo[j] - (cfo[j - 1] + cfo[j - 2] + cfo[j - 3])
capex[j] = capex[j] - (capex[j - 1] + capex[j - 2] + capex[j - 3])
quarter = 4
# 잉여현금흐름 : 영업활동의흐름 - 유형자산의 증가
fcf[j] = (cfo[j] - capex[j])
# 날짜 계산
date_day = self.__check_weekend(date_year, date_month, date_day)
date = date_year + date_month + str(date_day).zfill(2)
date_string = date_year + '-' + date_month + '-' + str(date_day).zfill(2)
# 각 분기별 마지막 영업일의 시가총액
market_cap_df = self.factor_data.stock.get_market_cap_by_date(date, date, stock_code)
try:
market_cap[j] = market_cap_df.loc[date_string]["시가총액"]
except KeyError:
print(market_cap_df)
market_cap[j] = 0
# 각 데이터들을 순서대로 리스트를 만들어서 record변수에 저장한다.
record = [stock_code, date_string, market_cap[j], quarter, current_assets[j], liabilities[j], equity[j],
total_assets[j],
revenue[j], grossProfit[j], income[j], net_income[j], cfo[j],
fcf[j]]
# 각 사업보고서별로 계산한 데이터들을 data 변수에 차곡차곡 넣는다.
data.append(record)
return data
extract.py - __calculate_indicator()
def __calculate_indicator(self, df):
"""
추출한 재무제표 데이터를 이용해서 팩터데이터 및 참고데이터들을 계산한다.
:param df:
:return Dataframe:
"""
df.sort_values(by=['종목코드', '연도'], inplace=True)
# print(df)
# 분기별 PER
df['분기 PER'] = np.nan
# 분기별 PBR
df['분기 PBR'] = np.nan
df['분기 PEG'] = np.nan
df['PGPR'] = np.nan
df['PSR'] = np.nan
df['GP/A'] = np.nan
df['POR'] = np.nan
df['PCR'] = np.nan
df['PFCR'] = np.nan
df['NCAV/MC'] = np.nan
df['분기 ROE'] = np.nan
three_indicators = ['매출액', '영업이익', '당기순이익']
three_indicators_status = ['매출액 상태', '영업이익 상태', '당기순이익 상태']
three_qoq_growth_indicators = ['QoQ 매출액 증가율', 'QoQ 영업이익 증가율', 'QoQ 당기순이익 증가율']
three_yoy_growth_indicators = ['YoY 매출액 증가율', 'YoY 영업이익 증가율', 'YoY 당기순이익 증가율']
df_temp = pd.DataFrame(columns=df.columns)
# 전체 데이터들 중에 종목코드만 추출해서 배열로 만듦.
corp_ticker = df.loc[:, ["종목코드"]].drop_duplicates().values.tolist()
for row in corp_ticker:
if row is None:
continue
print(f"Calculating {row[0]} factor indicators")
# 종목코드별로 반복문이 실행됨.
df_finance = df[df["종목코드"] == row[0]].reset_index()
# 종목이 가진 데이터길이 만큼 반복. 3부터 시작하는 이유는, 4개분기 데이터로 계산하는 데이터때문에.
# 과거 데이터를 참조해야하는데, 최초 3개 데이터까지는 참조할 데이터가 없음.
for i in range(3, len(df_finance)):
# PER : 시가총액 / 당기 순이익
df_finance.loc[i, "분기 PER"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3][three_indicators[2]] + df_finance.iloc[i - 2][three_indicators[2]] +
df_finance.iloc[i - 1][three_indicators[2]] + df_finance.iloc[i][three_indicators[2]])
# PSR : 시가총액 / 매출액
df_finance.loc[i, "PSR"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3][three_indicators[0]] + df_finance.iloc[i - 2][three_indicators[0]] +
df_finance.iloc[i - 1][three_indicators[0]] + df_finance.iloc[i][three_indicators[0]])
# PGPR : 시가총액 / 매출총이익
df_finance.loc[i, "PGPR"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3]['매출총이익'] + df_finance.iloc[i - 2]['매출총이익'] +
df_finance.iloc[i - 1]['매출총이익'] + df_finance.iloc[i]['매출총이익'])
# POR : 시가총액 / 영업이익
df_finance.loc[i, "POR"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3][three_indicators[1]] + df_finance.iloc[i - 2][three_indicators[1]] +
df_finance.iloc[i - 1][three_indicators[1]] + df_finance.iloc[i][three_indicators[1]])
# PCR : 시가총액 / 영업활동 현금흐름
df_finance.loc[i, "PCR"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3]['영업활동현금흐름'] + df_finance.iloc[i - 2]['영업활동현금흐름'] +
df_finance.iloc[i - 1]['영업활동현금흐름'] + df_finance.iloc[i]['영업활동현금흐름'])
# PFCR : 시가총액 / 잉여현금 흐름
df_finance.loc[i, "PFCR"] = df_finance.iloc[i]['시가총액'] / (
df_finance.iloc[i - 3]['잉여현금흐름'] + df_finance.iloc[i - 2]['잉여현금흐름'] +
df_finance.iloc[i - 1]['잉여현금흐름'] + df_finance.iloc[i]['잉여현금흐름'])
# ROE : 당기순이익 / 자본총계
df_finance.loc[i, "분기 ROE"] = ((df_finance.iloc[i - 3][three_indicators[2]] +
df_finance.iloc[i - 2][three_indicators[2]] +
df_finance.iloc[i - 1][three_indicators[2]] +
df_finance.iloc[i][three_indicators[2]]) /
df_finance.iloc[i]['자본총계']) * 100
# PBR : 시가총액 / 자본총계
df_finance["분기 PBR"] = df_finance['시가총액'] / df_finance['자본총계']
# GP/A : 최근 분기 매출총이익 / 자산총계
df_finance["GP/A"] = (df_finance['매출총이익'] / df_finance['자산총계']) * 100
# NCAV/MK : 청산가치(유동자산 - 부채총계) / 시가총액
df_finance["NCAV/MC"] = (df_finance['유동자산'] - df_finance['부채총계']) / \
df_finance['시가총액'] * 100
# 부채 비율
df_finance['부채비율'] = (df_finance['부채총계'] / df_finance['자본총계']) * 100
# 분기별 매출액 / 영업이익 / 당기순이익 증가율
for i in range(0, len(three_indicators)):
s = df_finance[three_indicators[i]].shift()
df_finance[three_qoq_growth_indicators[i]] = df_finance[three_indicators[i]].sub(s).div(s.abs()) * 100
# 전년동기대비 매출액 / 영업이익 / 당기순이익 증가율
for i in range(0, len(three_indicators)):
s = df_finance.groupby('분기')[three_indicators[i]].shift()
df_finance[three_yoy_growth_indicators[i]] = df_finance[three_indicators[i]].sub(s).div(s.abs()) * 100
# PEG : PER / 당기순이익 증가율
df_finance["분기 PEG"] = df_finance['분기 PER'] / df_finance[three_qoq_growth_indicators[2]]
# 매출총이익률 / 영업이익률 / 당기순이익률
df_finance['매출총이익률'] = (df_finance['매출총이익'] / df_finance[three_indicators[0]]) * 100
df_finance['영업이익률'] = (df_finance[three_indicators[1]] / df_finance[three_indicators[0]]) * 100
df_finance['당기순이익률'] = (df_finance[three_indicators[2]] / df_finance[three_indicators[0]]) * 100
# 분기열 삭제
df_finance.drop(['분기'], axis=1)
# 정렬 순서를 다시 바꿈. 과거 -> 현재순으로.
df_finance.sort_values(by=['연도'], inplace=True, ascending=False)
# 매출액, 영업이익, 당기순이익 확인 지표
# 이전 분기의 값과 비교하여 흑자인지 적자인지를 판단.
for i in range(len(three_indicators_status)):
df_finance[three_indicators_status[i]] = np.nan
df_finance.loc[
(df_finance[three_indicators[i]] > 0) & (df_finance[three_indicators[i]].shift(-1) <= 0),
three_indicators_status[i]
] = "흑자 전환"
df_finance.loc[
(df_finance[three_indicators[i]] <= 0) & (df_finance[three_indicators[i]].shift(-1) > 0),
three_indicators_status[i]
] = "적자 전환"
df_finance.loc[
(df_finance[three_indicators[i]] > 0) & (df_finance[three_indicators[i]].shift(-1) > 0),
three_indicators_status[i]
] = "흑자 지속"
df_finance.loc[
(df_finance[three_indicators[i]] <= 0) & (df_finance[three_indicators[i]].shift(-1) <= 0),
three_indicators_status[i]
] = "적자 지속"
## 기존 데이터프레임 하단에 종목별로 정제데이터들을 붙이기.
df_temp = pd.concat([df_finance, df_temp])
### 데이터들의 헤더 순서를 다시 지정해준다.
return df_temp.reindex(
columns=[
'종목코드', '연도', '시가총액',
'분기 PBR', 'PSR', 'PGPR', 'POR', '분기 PER', '분기 PEG', 'PCR', 'PFCR',
'분기 ROE', 'GP/A', 'NCAV/MC'
]
+ self.indicators +
[
'부채비율', '매출총이익률', three_qoq_growth_indicators[0], three_yoy_growth_indicators[0],
three_indicators_status[0],
'영업이익률', three_qoq_growth_indicators[1], three_yoy_growth_indicators[1],
three_indicators_status[1],
'당기순이익률', three_qoq_growth_indicators[2], three_yoy_growth_indicators[2],
three_indicators_status[2]
]
)필터 조건 추가 및 수정
재무제표와 팩터데이터를 구한 다음에는, filter_by_condition.py 의 필터 조건을 추가해봅니다.
저PER
저 PER 필터에도 조건을 추가합니다. 모든 데이터에 대해서도 필터링을 하고자하는데, 코스피, 코스닥 종목이 전부 들어있는 베이스가 되는 데이터는 재무제표가 포함되어있지 않아, 컬럼이 조금 다릅니다. 조건문을 추가해줍니다.
def filtering_low_per(sheet_name, df_copied: pd.DataFrame, all_data=False):
"""
전체 데이터중 조회시점을 기준으로 PER가 10 이하인 기업.
:param data:
:return:
"""
df = drop_column(df_copied)
df = df[df["PER"] > 0]
if all_data:
return (sheet_name,
df[df["PER"] <= 10.0].sort_values(by=["PER"], ascending=[True]))
else:
return (sheet_name,
df[df["PER"] <= 10.0].sort_values(by=["연도", "PER"], ascending=[False, True]))소형주 PEG
추가한 분기별 PEG 데이터를 이용해서 필터링을 해봅니다.
def filtering_peg(sheet_name, df: pd.DataFrame):
"""
PEG, PER/이익 성장률
최근분기 데이터로 계산
:param sheent_name:
:param df:
:return:
"""
df.drop(
df[df["부채비율"] >= 400.0].index,
inplace=True
)
df.drop(
df[df["분기 PEG"] <= 0].index,
inplace=True
)
peg_condition = (df['분기 PEG'] < 1.2)
df2 = df[peg_condition]
return (sheet_name,
df2.sort_values(by=['연도', '분기 PEG'], ascending=[False, True]).reset_index(
drop=True)
)팩터데이터 계산2
새롭게 추가한 데이터를 기반으로, 필터를 하나 추가합니다.
몇가지 조건들에 점수를 매기고, 토탈점수의 오름차순으로 정렬을 합니다.
사실 거의 다 때려넣었습니다.
def filtering_value_factor2(sheet_name, df: pd.DataFrame):
"""
아래의 높은순 랭크
- ROE, 영업이익률, 순이익률
아래 낮은 순 랭크
- 분기 PER, PBR, PCR, PSR, POR, PFCR, PGPR
분기 PER값을 사용
:param df:
:return:
"""
df.drop(
df[df["분기 PER"] <= 0].index,
inplace=True
)
df.drop(
df[df["PCR"] <= 0].index,
inplace=True
)
df.drop(
df[df["PSR"] <= 0].index,
inplace=True
)
df.drop(
df[df["POR"] <= 0].index,
inplace=True
)
df.drop(
df[df["PGPR"] <= 0].index,
inplace=True
)
df.drop(
df[df["YoY 매출액 증가율"] <= 0].index,
inplace=True
)
df.drop(
df[df["YoY 영업이익 증가율"] <= 0].index,
inplace=True
)
df.drop(
df[df["YoY 당기순이익 증가율"] <= 0].index,
inplace=True
)
df["ROE rank"] = df.groupby("연도")["분기 ROE"].rank(ascending=False)
df["영업이익률 rank"] = df.groupby("연도")["영업이익률"].rank(ascending=False)
df["순이익률 rank"] = df.groupby("연도")["당기순이익률"].rank(ascending=False)
df["YoY 영업이익 증가율 rank"] = df.groupby("연도")["YoY 영업이익 증가율"].rank(ascending=False)
df["YoY 당기순이익 증가율 rank"] = df.groupby("연도")["YoY 당기순이익 증가율"].rank(ascending=False)
df["PBR rank"] = df.groupby("연도")["PBR"].rank(ascending=True) # 기업 가치
df["PER rank"] = df.groupby("연도")["분기 PER"].rank(ascending=True) # 순자산
df["PCR rank"] = df.groupby("연도")["PCR"].rank(ascending=True) # 영업활동 현금흐름
df["PFCR rank"] = df.groupby("연도")["PFCR"].rank(ascending=True) # 잉여 현금흐름
df["PSR rank"] = df.groupby("연도")["PSR"].rank(ascending=True) # 매출
df["POR rank"] = df.groupby("연도")["POR"].rank(ascending=True) # 영업이익
df["PGPR rank"] = df.groupby("연도")["PGPR"].rank(ascending=True) # 메츨총이익
df["Total Value score"] = \
df["ROE rank"] + \
df["영업이익률 rank"] + \
df["순이익률 rank"] + \
df["PBR rank"] + \
df["PER rank"] + \
df["PCR rank"] + \
df["PSR rank"] + \
df["POR rank"] + \
df["PFCR rank"] + \
df["PGPR rank"] + \
df["YoY 영업이익 증가율 rank"] + \
df["YoY 당기순이익 증가율 rank"]
return (sheet_name,
df.sort_values(by=['연도', 'Total Value score'], ascending=[False, True]).reset_index(
drop=True)
)
팩터데이터 계산3
조금 다른 데이터들을 조합해서 다른 필터도 조합합니다.
def filtering_value_factor3(sheet_name, df: pd.DataFrame):
"""
PBR, PER, PGPR, PSR, POR의 합게 점수값이 낮은 순.
분기 PER값을 사용
:param df:
:return:
"""
df.drop(
df[df["분기 PER"] <= 0].index,
inplace=True
)
df.drop(
df[df["PCR"] <= 0].index,
inplace=True
)
df.drop(
df[df["PSR"] <= 0].index,
inplace=True
)
df.drop(
df[df["POR"] <= 0].index,
inplace=True
)
df.drop(
df[df["PGPR"] <= 0].index,
inplace=True
)
df.drop(
df[df["YoY 당기순이익 증가율"] <= 0].index,
inplace=True
)
df["PBR rank"] = df.groupby("연도")["PBR"].rank(ascending=True) # 기업 가치
df["PSR rank"] = df.groupby("연도")["PSR"].rank(ascending=True) # 매출
df["PGPR rank"] = df.groupby("연도")["PGPR"].rank(ascending=True) # 매출총이익
df["POR rank"] = df.groupby("연도")["PGPR"].rank(ascending=True) # 영업이익
df["PER rank"] = df.groupby("연도")["분기 PER"].rank(ascending=True) # 순자산
df["PEG rank"] = df.groupby("연도")["분기 PEG"].rank(ascending=True) # 순자산의 성장률
df["Total Value score"] = df["PBR rank"] + df["PSR rank"] + df["PGPR rank"] + df["POR rank"] + df["PER rank"] + df[
"PEG rank"]
return (sheet_name,
df.sort_values(by=['연도', 'Total Value score'], ascending=[False, True]).reset_index(
drop=True)
)filter_by_condition.py 는 위 외에도, 원래 ~ 증가율 로 되어있던 몇몇 함수에 대해 QoQ ~ 증가율 로 고쳐주었습니다. (YoY, QoQ 를 명시하는 것으로 리팩토링을 했기 때문)
또한 몇몇 정렬 기준에 대해서도 조금 수정을 해주었습니다.
파일의 모든 내용을 기재하기에는 포스트가 길어지므로, 자세한 내용은 아래 레포지토리의 파일내용을 확인해주세요.
포스트가 길어져, 이번에 추가한 팩터데이터에 대한 검증은 다음 포스트에서 기재해봅니다.
ref
반응형
'파이썬으로 종목 스크리너 만들기' 카테고리의 다른 글
| 뇌동매매 금지 - 11. S-RIM을 이용한 목표주가 계산해보기 - 1. S-RIM 개념 알아보기 (0) | 2022.11.26 |
|---|---|
| 뇌동매매 금지 - 10. Bug fix와 이것저것 추가하기 - 3 (1) | 2022.09.12 |
| 뇌동매매 금지 - 10. Bug fix와 이것저것 추가하기 - 1 (0) | 2022.09.12 |
| 뇌동매매 금지 - 9. 스크리너 제작 후기 (0) | 2022.09.12 |
| 뇌동매매 금지 - 8. 상장종목 스크리닝 해보기 (0) | 2022.09.12 |
댓글