ポートフォリオ可視化スクリプト(4) - 投資適格判断のスクリーニング -¶
コード¶
stock_toolkit.py¶
import os, sys, shutil, json5, datetime, tqdm, re, time, requests
import concurrent.futures
import jinja2
import itertools as it
import yfinance as yf
import numpy as np
import pandas as pd
import nk_toolkit.math.round__digit as rdg
sensitivesList = [ "鉱業", "石油・石炭製品", "ガラス・土石製品", "鉄鋼", "非鉄金属", "金属製品",
"化学", "ゴム製品", "パルプ・紙", "繊維製品", "機械", "輸送用機器", "電気機器",
"精密機器", "その他製品", "陸運業", "海運業", "空運業", "倉庫・運輸関連業",
"建設業", "不動産業", "卸売業" ]
defensivesList = [ "食料品", "医薬品", "電気・ガス業", "小売業", "銀行業", "保険業", \
"証券、商品先物取引業", "その他金融業", "サービス業", "情報・通信業" ]
othersList = [ "-" ]
# ========================================================= #
# === analyze__stockStatus.py === #
# ========================================================= #
def analyze__stockStatus( settingFile="dat/settings.json" ):
# ------------------------------------------------- #
# --- [1] load settings.json --- #
# ------------------------------------------------- #
with open( settingFile, "r" ) as f:
params = ( json5.load( f ) )
settings = params["settings"]
nyears = settings["nyears"]
# ------------------------------------------------- #
# --- [2] load previous database --- #
# ------------------------------------------------- #
try:
with open( settings["databaseFile"], "r" ) as f:
data_ = json5.load( f )
except:
print( "[analyze__stockStatus.py] no databaseFile :: {} "\
.format( settings["databaseFile"] ) )
data_ = {}
# ------------------------------------------------- #
# --- [3] prepare ticker from list --- #
# ------------------------------------------------- #
if ( settings["mode"] in ["all"] ):
jpx_list = pd.read_csv( settings["filteredFile"] )
candidate = settings["markets"]
params["tickers"] = jpx_list[ jpx_list["市場・商品区分"].isin( candidate ) ]["コード"]
elif ( settings["mode"] in [ "select" ] ):
pass
else:
print( "[stock_toolkit.py] unrecognizable mode ... {} ".format( params["mode"] ) )
sys.exit()
if ( settings["tickersFile"] is not None ):
stocklist = pd.read_csv( settings["tickersFile"] )
name_map = dict(zip(stocklist["コード"], stocklist["銘柄名"]))
sector_map = dict(zip(stocklist["コード"], stocklist["33業種区分"]))
stocklist = { "name":name_map, "sector":sector_map }
# ------------------------------------------------- #
# --- [4] skip already stored in database --- #
# ------------------------------------------------- #
skips = []
fmt = "%Y/%m/%d"
today = datetime.datetime.today().strftime('%Y/%m/%d')
if ( settings["skip"] ):
for ik,tic in enumerate( params["tickers"] ):
if ( tic in data_ ):
date_db = datetime.datetime.strptime( data_[tic]["date"], fmt ).date()
date_nw = datetime.datetime.strptime( today, fmt ).date()
if ( date_db >= date_nw ):
print( "[stock_toolkit.py] ticker = {}'s data is up-to-date... [skip]"\
.format( tic ) )
skips += [ tic ]
params["tickers"] = [ tic for tic in params["tickers"] if tic not in skips ]
# ------------------------------------------------- #
# --- [5] seek info of a stock --- #
# ------------------------------------------------- #
stack = []
def workers( ticker, nyears, stocklist, params ):
ret = seek__stockinfo( ticker=ticker, \
nyears=nyears, stocklist=stocklist, params=params )
return( ret )
with concurrent.futures.ThreadPoolExecutor( max_workers=settings["nParallels"] ) as ex:
futures = [ ex.submit( workers, tic, nyears, stocklist, params ) for tic in params["tickers"] ]
for fu in tqdm.tqdm( concurrent.futures.as_completed( futures ), total=len(futures) ):
stack += [ fu.result() ]
rets = { stock["ticker"]:stock for stock in stack }
data = { **data_, **rets }
# ------------------------------------------------- #
# --- [6] own flags / sector info --- #
# ------------------------------------------------- #
for tic in params["owns"]:
if ( tic in data ):
data[tic]["own"] = True
else:
print( "[stock_toolkit.py] no database data of {}.. cannot add own flag.. [WARNING]"\
.format( tic ) )
with open( settings["databaseFile"], "w", encoding="utf-8" ) as f:
json5.dump( data, f, indent=4, quote_keys=True )
return( data )
# tickers = [ "{}.T".format( tic ) for tic in params["tickers"] ]
# tickers = tickers[:100]
# tickers = yf.Tickers( tickers )
# stack = {}
# import tqdm.contrib.concurrent
# import functools
# func = functools.partial( seek__stockinfo, nyears=nyears, rename=rename, params=params )
# rets = tqdm.contrib.concurrent.thread_map( func, params["tickers"], \
# max_workers=settings["nParallels"] )
# stack = { key:rets[ik] for ik,key in enumerate(params["tickers"]) }
# with concurrent.futures.ThreadPoolExecutor( max_workers=settings["nParallels"] ) as ex:
# rets = ex.map( workers, params["tickers"], \
# it.repeat( nyears ), it.repeat( rename ), it.repeat( params ) )
# rets = list( rets )
# stack = { key:rets[ik] for ik,key in enumerate(params["tickers"]) }
# futures = [ ex.submit( workers, tickers, tic, nyears, rename, params ) \
# for tic in params["tickers"] ]
# for fu in tqdm.tqdm( concurrent.futures.as_completed( futures ), total=len(futures) ):
# stack[tic] = fu.()
# for ik,tic in enumerate( tqdm.tqdm( params["tickers"] ) ):
# stack[tic] = seek__stockinfo( tickers=tickers, ticker=tic, \
# nyears=nyears, rename=rename, params=params )
# if ( ik%20 == 0 ): # -- 途中セーブ -- #
# data = { **data_, **stack }
# with open( settings["databaseFile"], "w", encoding="utf-8" ) as f:
# json5.dump( data, f, indent=4, quote_keys=True )
# ========================================================= #
# === make HTML file for display === #
# ========================================================= #
def makehtml__stockStatus( settingFile="dat/settings.json" ):
# ------------------------------------------------- #
# --- [1] load settings.json / database.json --- #
# ------------------------------------------------- #
with open( settingFile, "r" ) as f:
params = ( json5.load( f ) )
settings = params["settings"]
# ------------------------------------------------- #
# --- [2] set keys --- #
# ------------------------------------------------- #
keys = {}
for key in params["rangecheck"].keys():
rmin = ( "{0}".format( params["rangecheck"][key]["min"] ) ).replace( "None", " " )
rmax = ( "{0}".format( params["rangecheck"][key]["max"] ) ).replace( "None", " " )
keys[key] = params["rangecheck"][key]["name"] + " ( {0} - {1} )".format( rmin, rmax )
for key in params["increment"].keys():
keys[key] = params["increment"][key]["name"]
# ------------------------------------------------- #
# --- [3] load previous database --- #
# ------------------------------------------------- #
try:
with open( settings["databaseFile"], "r" ) as f:
data_ = json5.load( f )
except:
print( "[analyze__stockStatus.py] no databaseFile :: {} "\
.format( settings["databaseFile"] ) )
data_ = {}
# ------------------------------------------------- #
# --- [4] sort by score --- #
# ------------------------------------------------- #
if ( settings["sort"] ):
tickers = pd.Series( [ key for key in data_.keys() ] )
scores = pd.Series( [ data_[key]["score"] for key in data_.keys() ] )
df = pd.DataFrame( { "ticker":tickers, "score":scores } )
df = df.sort_values( "score", ascending=False )
sortkeys = list( df["ticker"] )
data_ = { key:data_[key] for key in sortkeys }
data = [ data_[key] for key in data_.keys() ]
# ------------------------------------------------- #
# --- [5] render html file with jinja2 --- #
# ------------------------------------------------- #
env = jinja2.Environment( loader=jinja2.FileSystemLoader("templates") )
template = env.get_template( settings["template.detail"] )
html_output = template.render( data=data, keys=keys )
shutil.copy( settings["cssFile"], "html/" )
with open( settings["html.fancy"], "w", encoding="utf-8" ) as f:
f.write( html_output )
print(" output :: {}".format( settings["html.fancy"] ) )
# ------------------------------------------------- #
# --- [6] render screening table using jinja2 --- #
# ------------------------------------------------- #
today = datetime.datetime.today().strftime("%Y-%m-%d")
env = jinja2.Environment( loader=jinja2.FileSystemLoader("templates") )
template = env.get_template( settings["template.table"] )
html_output = template.render( data=data_, keys=keys, today=today )
shutil.copy( settings["cssFile"], "html/" )
with open( settings["html.screening"], "w", encoding="utf-8" ) as f:
f.write( html_output )
print(" output :: {}".format( settings["html.screening"] ) )
return()
# ========================================================= #
# === seek__stockinfo.py === #
# ========================================================= #
def seek__stockinfo( tickers=None, ticker=None, stocklist=None, digit=3, nyears=15, params=None ):
# ------------------------------------------------- #
# --- [1] 財務データ取得 --- #
# ------------------------------------------------- #
if ( tickers is None ):
hticker = yf.Ticker( ticker+".T" )
else:
hticker = tickers.tickers[ ticker+".T" ]
financials = hticker.financials
balance_sheet = hticker.balance_sheet
cashflow = hticker.cashflow
fast_info = hticker.fast_info
bs_items = balance_sheet.index
pl_items = financials.index
cf_items = cashflow.index
results = {}
# ------------------------------------------------- #
# --- [2] 基本情報 --- #
# ------------------------------------------------- #
results["ticker"] = ticker
results["name"] = ticker
results["date"] = datetime.datetime.today().strftime('%Y/%m/%d')
results["own"] = False
price = hticker.fast_info.last_price
shares = hticker.get_shares_full()
net_income = np.nan
equity = np.nan
total_assets = np.nan
eps = np.nan
bps = np.nan
if shares is not None:
shares = shares.sort_index().iloc[-1]
else:
shares = balance_sheet.loc["Ordinary Shares Number"]
if ( shares is not None ) and ( len(shares) > 0 ):
shares = shares.sort_index().iloc[0]
if ( "Net Income" in pl_items ):
net_income = financials.loc["Net Income"]
if ( "Common Stock Equity" in bs_items ):
equity = balance_sheet.loc["Common Stock Equity"].iloc[0]
if ( "Total Assets" in bs_items ):
total_assets = balance_sheet.loc["Total Assets"].iloc[0]
results["currentPrice"] = price
if ( "Net Income Common Stockholders" in hticker.ttm_income_stmt.index ):
ni_ttm = float( hticker.ttm_income_stmt.loc["Net Income Common Stockholders"].iloc[0] )
elif ( "Net Income" in hticker.ttm_income_stmt.index ):
ni_ttm = hticker.ttm_income_stmt.loc["Net Income"].iloc[0]
elif ( "Net Income Common Stockholders" in hticker.income_stmt.index ):
ni_ttm = float( hticker.income_stmt.loc["Net Income Common Stockholders"].iloc[0] )
elif ( "Net Income" in hticker.income_stmt.index ):
ni_ttm = hticker.income_stmt.loc["Net Income"].iloc[0]
elif ( np.isfinite( net_income ).any() ):
ni_ttm = net_income.iloc[0]
else:
ni_ttm = np.nan
if ( np.isfinite( ni_ttm ) and np.isfinite( shares ) ):
eps = ni_ttm / shares
if ( np.isfinite( equity ) and np.isfinite( shares ) ):
bps = equity / shares
# ------------------------------------------------- #
# --- [3] 売上高推移 --- #
# ------------------------------------------------- #
# -- [3-1] 売上高推移 -- #
try:
results["revenues"] = list( ( financials.loc["Total Revenue"] ).to_numpy() )
except KeyError:
results["revenues"] = np.nan
# -- [3-2] 営業利益率 -- #
try:
operating_income = financials.loc["Operating Income"]
results["oprIncome"] = operating_income.iloc[0] / results["revenues"][0] * 100.0 # [%]
except Exception:
results["oprIncome"] = np.nan
# ------------------------------------------------- #
# --- [4] 売上高・経常利益・純利益の推移 --- #
# ------------------------------------------------- #
results["revenues.inc"] = pd.Series([], dtype="float64")
results["dividend.inc"] = pd.Series([], dtype="float64")
results["oprIncome.inc"] = pd.Series([], dtype="float64")
results["netIncome.inc"] = pd.Series([], dtype="float64")
results["EPS.inc"] = pd.Series([], dtype="float64")
dividend = hticker.dividends
dividend = correct__latest_dividend( dividend )
if ( not( dividend.empty ) ):
results["dividend.inc"] = pd.Series( dividend["dividend"].values )
if ( "Total Revenue" in pl_items ):
results["revenues.inc"] = financials.loc["Total Revenue"] / 1.0e8
if ( "Operating Income" in pl_items ):
results["oprIncome.inc"] = operating_income / 1.0e8
if ( "Net Income" in pl_items ):
results["netIncome.inc"] = net_income / 1.0e8
if ( "Net Income" in pl_items ) and ( "Ordinary Shares Number" in bs_items ):
results["EPS.inc"] = net_income / balance_sheet.loc["Ordinary Shares Number"]
elif ( "Share Issued" in bs_items ) and ( np.isfinite(results["netIncome.inc"] ).any() ):
results["EPS.inc"] = results["netIncome.inc"] / balance_sheet.loc["Share Issued"]
for key in ["oprIncome.inc", "revenues.inc", "netIncome.inc", "EPS.inc", "dividend.inc" ]:
values = ( ( (results[key].sort_index()).dropna() ).values ).astype( int )
results[key] = list( [ int(val) for val in values[::-1][:nyears][::-1] ] )
# try:
# values = ( ( results[key].dropna() ).values ).astype( int )
# results[key] = list( [ int(val) for val in values[::-1] ] )
# except:
# results[key] = np.nan
if ( np.isnan( eps ) and len( results["EPS.inc"] ) > 0 ):
eps = results["EPS.inc"][-1]
# ------------------------------------------------- #
# --- [5] 配当利回り, 配当性向 --- #
# ------------------------------------------------- #
results["dividend.yield"] = np.nan
results["dividend.ratio"] = np.nan
if ( not( dividend.empty ) and ( np.isfinite(price) ) ):
results["dividend.yield"] = dividend["dividend"].iloc[-1] / price * 100.0
if ( not( dividend.empty ) and ( len( results["EPS.inc"] ) > 0 ) ):
results["dividend.ratio"] = dividend["dividend"].iloc[-1] / eps * 100.0
# ------------------------------------------------- #
# --- [6] EPS / PER / PBR --- #
# ------------------------------------------------- #
results["EPS"] = np.nan
results["PER"] = np.nan
results["PBR"] = np.nan
if np.isfinite( eps ):
results["EPS"] = eps
if np.isfinite( eps ) and np.isfinite( price ):
results["PER"] = price / eps
if np.isfinite( bps ) and np.isfinite( price ):
results["PBR"] = price / bps
# ------------------------------------------------- #
# --- [7] ROE / ROA --- #
# ------------------------------------------------- #
results["ROE"] = np.nan
results["ROA"] = np.nan
if ( np.isfinite( ni_ttm ) and np.isfinite( equity ) ):
results["ROE"] = ni_ttm / equity * 100.0 # [%]
if ( np.isfinite( ni_ttm ) and np.isfinite( total_assets ) ):
results["ROA"] = ni_ttm / total_assets * 100.0 # [%]
# ------------------------------------------------- #
# --- [8] 自己資本比率 / 流動比率 / 現金比率 --- #
# ------------------------------------------------- #
# -- [8-1] 自己資本比率 -- #
try:
results["equityRatio"] = equity / total_assets * 100.0 # [%]
except:
results["equityRatio"] = np.nan
# -- [8-2] 流動比率 -- #
try:
current_assets = balance_sheet.loc["Current Assets"].iloc[0]
current_liabilities = balance_sheet.loc["Current Liabilities"].iloc[0]
results["currentRatio"] = current_assets / current_liabilities * 100.0 # [%]
except Exception:
results["currentRatio"] = np.nan
# -- [8-3] 現金比率 -- #
try:
cash = balance_sheet.loc["Cash And Cash Equivalents"].iloc[0]
results["cashRatio"] = cash / total_assets * 100.0 # [%]
except Exception:
results["cashRatio"] = np.nan
# ------------------------------------------------- #
# --- [9] 有利子負債比率 --- #
# ------------------------------------------------- #
# -- income / depreciation :: denominator -- #
if ( ( "Operating Income" in pl_items ) and ( "Depreciation" in cf_items ) ):
oprat_income = financials.loc["Operating Income"].iloc[0]
depreciation = cashflow.loc["Depreciation"].iloc[0]
ebitda = oprat_income + depreciation
else:
ebitda = None
# -- debt -- #
if ( "Total Debt" in bs_items ):
debt = balance_sheet.loc["Total Debt"].iloc[0]
elif ( "Short Long Term Debt" in bs_items ):
debt = balance_sheet.loc["Short Long Term Debt"].iloc[0]
elif ( "Short Term Debt" in bs_items ) and ( "Long Term Debt" in bs_items ):
debt = balance_sheet.loc["Short Term Debt"].iloc[0] + \
balance_sheet.loc["Long Term Debt"].iloc[0]
else:
debt = None
# -- cash -- #
if ( "Cash And Cash Equivalents" in bs_items ):
cash = balance_sheet.loc["Cash And Cash Equivalents"].iloc[0]
else:
cash = None
# -- debtRatio -- #
if ( ebitda and debt and cash ):
results["debtRatio.ebitda"] = ( debt - cash ) / ebitda
else:
results["debtRatio.ebitda"] = np.nan
if ( debt and equity ):
results["debtRatio.equity"] = debt / equity * 100.0
else:
results["debtRatio.equity"] = np.nan
# ------------------------------------------------- #
# --- [10] return --- #
# ------------------------------------------------- #
results["check"] = check__stockinfo( stock=results, params=params )
results["score"] = int( np.sum( [ int(val) for val in results["check"].values() ] ) )
results["sensitive"] = False
max_score = len( results["check"].values() )
if ( stocklist is not None ):
results["name"] = stocklist["name"] .get( results["ticker"], results["name"] )
results["sector"] = stocklist["sector"].get( results["ticker"], "unknown" )
max_score = max_score + 1
if ( results["sector"] in sensitivesList ):
results["sensitive"] = True
else:
results["sensitive"] = False
results["score"] += 1
results["score_display"] = "{0}/{1}".format( results["score"], max_score )
results["score_bgcolor"] = score_to_color( results["score"], max_score )
if ( digit ):
import numbers
for key in results.keys():
if ( ( isinstance(results[key], numbers.Number) ) and not( np.isnan(results[key] ) ) ):
results[key] = rdg.round__digit( results[key], digit=digit )
for key in results.keys():
if ( isinstance( results[key], ( float, np.floating ) ) ):
if ( pd.isna( results[key] ) ): results[key] = "N/A"
return( results )
# TTM = info.get( "trailingAnnualDividendRate", np.nan )
# yfina_div = info.get( "dividendRate", np.nan )
# if ( results["dividend.inc"] != np.nan ): # -- 直近配当が中間配当までの場合を補正 -- #
# latest = results["dividend.inc"][-1]
# else:
# latest = np.nan
# if ( latest == TTM ) or ( latest == yfina_div ):
# results["dividend.yield"] = latest / price * 100.0
# elif ( TTM == yfina_div ):
# results["dividend.yield"] = TTM / price * 100.0
# else:
# vals = np.array( [ latest, TTM, yfina_div ] )
# div = np.median( vals[ ~np.isnan(vals) ] )
# results["dividend.yield"] = div / price * 100.0
# if ( TTM is not None ):
# if ( results["dividend.inc"] != np.nan ): # -- 直近配当が中間配当までの場合を補正 -- #
# latest_dividend = results["dividend.inc"][-1]
# if ( latest_dividend != TTM ):
# try:
# if ( results["dividend.inc"][-1] <= 0.5*results["dividend.inc"][-2] ):
# results["dividend.inc"][-1] = int(TTM)
# results["dividend.yield"] = int(TTM) / results["currentPrice"] * 100.0
# except:
# try:
# if ( results["dividend.inc"][-1] <= 0.5*results["dividend.inc"][-2] ):
# results["dividend.inc"].pop( -1 )
# try:
# results["dividend.yield"] = results["dividend.inc"][-1] / results["currentPrice"] * 100.0
# except:
# try:
# results["dividend.yield"] = info.get( "dividendRate", np.nan ) / results["currentPrice"] * 100.0
# except:
# results["dividend.yield"] = None
# except:
# pass
# ------------------------------------------------- #
# --- [8] 株価推移と移動平均線 --- #
# ------------------------------------------------- #
# try:
# history = hticker.history(period="5y")
# results["stock_price"] = history["Close"]
# results["ma50"] = history["Close"].rolling(window=50).mean()
# except:
# results["stock_price"] = np.nan
# results["ma50"] = np.nan
# ========================================================= #
# === check__stockinfo.py === #
# ========================================================= #
def check__stockinfo( stock=None, settingFile="dat/settings.json", params=None ):
# ------------------------------------------------- #
# --- [1] load json settings --- #
# ------------------------------------------------- #
if ( params is None ):
with open( settingFile, "r" ) as f:
params = json5.load( f )
rangecheck = params["rangecheck"]
increment = params["increment"]
# ------------------------------------------------- #
# --- [2] range check --- #
# ------------------------------------------------- #
ret = {}
for key in rangecheck.keys():
ret[key] = True
if ( rangecheck[key]["min"] is not None ):
try:
if ( stock[key] <= rangecheck[key]["min"] ): ret[key] = False
except:
ret[key] = False
if ( rangecheck[key]["max"] is not None ):
try:
if ( stock[key] >= rangecheck[key]["max"] ): ret[key] = False
except:
ret[key] = False
# ------------------------------------------------- #
# --- [3] increment check --- #
# ------------------------------------------------- #
for key in increment.keys():
try:
values = np.array( [ float( val ) for val in stock[key] ] )
ret[key] = bool( np.all( np.diff( values ) >= 0.0 ) )
except:
ret[key] = False
# ------------------------------------------------- #
# --- [4] sector check --- #
# ------------------------------------------------- #
return( ret )
# ========================================================= #
# === score to color function === #
# ========================================================= #
def score_to_color( score, max_score=18 ):
ratio = max(0, min(1, score / max_score))
if ratio <= 0.5: # 青→白
t = ratio / 0.5
r = int(255 * t)
g = int(255 * t)
b = 255
else: # 白→オレンジ
t = (ratio - 0.5) / 0.5
r = 255
g = int(255 * (1 - 0.5 * t))
b = int(255 * (1 - t))
return f"rgb({r},{g},{b})"
# ========================================================= #
# === filter__jpx_stock_list === #
# ========================================================= #
def filter__jpx_stock_list( settingFile="dat/settings.json" ):
# ------------------------------------------------- #
# --- [1] load jpx list --- #
# ------------------------------------------------- #
with open( settingFile, "r" ) as f:
params = ( json5.load( f ) )
settings = params["settings"]
jpx_list = pd.read_csv( settings["tickersFile"] )
min_yield = settings["div.yield.min"]
candidate = settings["markets"]
tickerList = jpx_list[ jpx_list["市場・商品区分"].isin( candidate ) ]["コード"].to_list()
tickerList_ = " ".join( [ tic + ".T" for tic in tickerList ] )
yftickers = yf.Tickers( tickerList_ )
# ------------------------------------------------- #
# --- [2] calculate dividend yield --- #
# ------------------------------------------------- #
def calc__dividend_yield( tic, yftickers, min_yield ):
stock = yftickers.tickers[tic+".T"]
price = stock.fast_info.last_price
divs = stock.dividends
dividend = correct__latest_dividend( divs )
if ( divs.empty or not( np.isfinite( price ) ) ):
return None
else:
div_yield = dividend["dividend"].iloc[-1] / price * 100.0
if ( div_yield > min_yield ):
return tic
else:
return None
# ------------------------------------------------- #
# --- [3] calculate dividend yieldrate --- #
# ------------------------------------------------- #
stack = []
with concurrent.futures.ThreadPoolExecutor( max_workers=settings["nParallels"] ) as ex:
futures = [ ex.submit( calc__dividend_yield, tic, yftickers, min_yield ) \
for tic in tickerList]
for fu in tqdm.tqdm( concurrent.futures.as_completed( futures ), total=len(futures) ):
ret = fu.result()
if ( ret is not None ):
stack += [ ret ]
filtered = jpx_list[ jpx_list["コード"].isin(stack) ]
# ------------------------------------------------- #
# --- [4] save in a file --- #
# ------------------------------------------------- #
filtered.to_csv( settings["filteredFile"] )
return( filtered )
# stack = []
# for tic in tqdm.tqdm(tickerList):
# price = np.nan
# div_yield = np.nan
# price = yftickers.tickers[ tic+".T" ].fast_info.last_price
# dividend_ = yftickers.tickers[ tic+".T" ].dividends
# dividend = correct__latest_dividend( dividend_ )
# if ( not( dividend.empty ) and ( np.isfinite( price ) ) ):
# div_yield = dividend["dividend"].iloc[-1] / price * 100.0
# if ( div_yield > settings["div.yield.min"] ):
# stack += [ tic ]
# filtered = jpx_list[ jpx_list["コード"].isin(stack) ]
# ========================================================= #
# === correct__dividends === #
# ========================================================= #
def correct__latest_dividend( divs: pd.Series, threshold=0.70, \
days=365, tz="Asia/Tokyo") -> pd.DataFrame:
# ------------------------------------------------- #
# --- [1] vacant / None input --- #
# ------------------------------------------------- #
if ( divs is None ) or ( divs.empty ):
return pd.DataFrame( columns=["date", "配当額"] )
# ------------------------------------------------- #
# --- [2] Date settings / calc annual dividend --- #
# ------------------------------------------------- #
s = pd.Series( divs ).dropna().sort_index()
s.index = pd.to_datetime( s.index )
s.index = s.index.tz_localize(tz) if s.index.tz is None else s.index.tz_convert(tz)
annual = s.resample("YE-DEC").sum().to_frame(name="dividend")
# ------------------------------------------------- #
# --- [3] check latest dividend --- #
# ------------------------------------------------- #
if len(annual) >= 2:
last_val = float( annual["dividend"].iloc[-1] )
prev_val = float( annual["dividend"].iloc[-2] )
if ( prev_val > 0 ) and ( last_val <= threshold*prev_val ):
ttm_cut = s.index.max() - pd.Timedelta( days=days )
ttm_sum = float( s[s.index > ttm_cut ].sum() )
annual.iloc[ -1, annual.columns.get_loc("dividend")] = ttm_sum
# ------------------------------------------------- #
# --- [4] return --- #
# ------------------------------------------------- #
ret = annual.reset_index().rename( columns={"index": "Date"} )
ret["Date"] = ret["Date"].apply(lambda x: x.replace(hour=0, minute=0, second=0, microsecond=0).isoformat(sep=" "))
return( ret[ ["Date", "dividend"] ] )
# ========================================================= #
# === Execution of Pragram === #
# ========================================================= #
if ( __name__=="__main__" ):
analyze__stockStatus()
makehtml__stockStatus()
settings.json¶
{
"rangecheck": {
"dividend.yield" : { "name": "配当利回り (%)" , "min":3.5, "max":5.5, },
"dividend.ratio" : { "name": "配当性向 (%)" , "min":10.0, "max":50.0, },
"oprIncome" : { "name": "営業利益率 (%)" , "min":10.0, "max":null, },
"PER" : { "name": "PER" , "min":5.0, "max":15.0, },
"PBR" : { "name": "PBR" , "min":0.5, "max":1.5, },
"ROE" : { "name": "ROE (%)" , "min":8.0, "max":null, },
"ROA" : { "name": "ROA (%)" , "min":5.0, "max":null, },
"equityRatio" : { "name": "自己資本比率 (%)" , "min":50.0, "max":null, },
"currentRatio" : { "name": "流動比率 (%)" , "min":200.0, "max":null, },
"cashRatio" : { "name": "現金比率 (%)" , "min":30.0, "max":null, },
"debtRatio.ebitda" : { "name": "EBITDA有利子負債倍率(倍)" , "min":null, "max":3.0, },
"debtRatio.equity" : { "name": "有利子負債倍率 (%)" , "min":null, "max":200, },
},
"increment": {
"revenues.inc" : { "name":"売上高 (億円)" },
"oprIncome.inc" : { "name":"営業利益 (億円)" },
"netIncome.inc" : { "name":"純利益 (億円)" },
"EPS.inc" : { "name":"EPS (円)" },
"dividend.inc" : { "name":"１株配当 (円)" },
},
// "tickers": [
// "8593",
// ],
"tickers": [
"2169", "2267", "2914", "3076", "3479", "3763", "3817", "4008",
"4752", "4928", "5020", "5108", "5192", "5334", "5388", "5401",
"5464", "5589", "6113", "6247", "6268", "6301", "7247", "7820",
"7921", "7994", "7995", "8002", "8053", "8058", "8316", "8411",
"8424", "8425", "8439", "8584", "8593", "8898", "9432", "9433",
"9513", "9795", "9799", "9986", "1605", "6501",
],
"owns": [
"2169", "2267", "2914", "3076", "3479", "3763", "3817", "4008",
"4752", "4928", "5020", "5108", "5192", "5334", "5388", "5401",
"5464", "5589", "6113", "6268", "6301", "7247", "7820",
"7921", "7994", "7995", "8002", "8053", "8058", "8316", "8411",
"8424", "8425", "8439", "8584", "8593", "8898", "9432", "9433",
"9513", "9795", "9799", "9986", "1605", "6501",
],
"settings":{
"mode" : "all" , // "all", "select"
"skip" : false , // whether skip or not.
"div.yield.min" : 3.0 , // minimum dividend rate
"cssFile" : "templates/custom.css" ,
"databaseFile" : "dat/database.json" ,
"tickersFile" : "dat/jpx_tickers.csv" ,
"filteredFile" : "dat/filtered_tickers.csv" ,
"markets" : [ "プライム（内国株式）", "スタンダード（内国株式）" ] ,
// [ "プライム（内国株式）", "スタンダード（内国株式）", "グロース（内国株式）" ],
"html.fancy" : "html/fancy_table.html" ,
"html.screening" : "html/screening_table.html" ,
"template.detail" : "fancy_table.j2.html" ,
"template.table" : "screening_table.j2.html" ,
"nyears" : 15 ,
"nParallels" : 4 ,
"sort" : true ,
},
}
留意事項¶
本コードはyfinanceを使用しています．
免責事項¶
Note
本コードは、オープンソースライブラリ yfinance を利用して Yahoo! Finance から取得可能な金融データを参照する例を示したものです。
本コードは学習・研究・個人利用を目的としたものであり、投資助言や売買推奨を行うものではありません。
本コードが取得するデータは Yahoo! Finance に依存しており、データの正確性、完全性、最新性を保証するものではありません。
Yahoo! Finance のデータ利用については、各自で[Yahoo!利用規約](https://legal.yahoo.com/) をご確認ください。
本コードを利用した結果生じたいかなる損害・損失についても、作者は一切の責任を負いません。
公開されているコードは自由に改変・利用可能ですが、 取得したデータの再配布や商用利用については利用規約を遵守してください 。
本コードを利用する際は、必ず自己責任でご利用ください。
