決算発表日情報をDBに保存する
たまにはプログラムについての投稿仕事とプライベートでのプログラムを書いてて違いをよく思うんですが、プライベートの場合は、影響範囲をそんなに考えなくてよいところと、テストをデバッグぐらいしかしないため開発スピードが早いです。あたりまえですがｗ言語も仕事で使うjavaとかC#よりもpythonのほうが開発スピードが早い気がします。今まで、業務（証券系）で仕事を選んでたけど、pythonの仕事もしてみたいな。あと平日少しずつ自分用の取引用WEB画面作成中です。デザインを考えてる最中・・むずかしい！しかし自分のためだけのWEBだから自由すぎて楽しい本題ですが、python3を使って東証の決算発表Excelをダウンロードして、pandasでExcelを読み込んで、決算発表情報をsqliteのテーブルに保存する方法やっていることは、①東証決算発表予定日のページにアクセス②ページを解析して、xlsのリンクを見つけてループする③xlsをローカルにダウンロードする④ダウンロードしたxlsをpandasで読み込みオブジェクト化する⑤DBに入れるためにデータを色々整える⑥整え終わったらDataFrameのリストに追加⑦ループ終了後、DataFrameのリストを一つのDataFrameに合体⑧一つのDataFrameをDictionary Listに変換する⑨データベースに保存する初めてpandas使ったんですがものすごい便利です。珍しくソースにコメントを書いたのでちょっと見やすいし（自分的には）ブログにでものっけるかと思い書いてたんですが、ものすごい見ずらい・・・あと、別ファイルにあるUtil系もとりあえずのせてますが、やっぱりみずらい・・・・・・途中まで書いてしまったので突っ走ります・・多分・・・最初で最後・・・ソースコード# 決算発表情報を取得しますimport sqlite3_dao as daoimport html_utilimport reimport pandas as pdimport calendar_util as cal# 決算発表情報を取得して保存しますdef save_announce_financial(): url = 'https://www.jpx.co.jp/listing/event-schedules/financial-announcement/index.html' # ①東証決算発表予定日のページにアクセス soup = html_util.request_beautiful_soup(url) df_list = [] today = datetime.now().strftime('%Y-%m-%d') # 全レコード省略せずにprintする pd.set_option("display.max_columns", 100) pd.set_option("display.max_rows", 2000) # ②ページを解析して、xlsのリンクを見つけてループする for href in soup.find_all('a', attrs={"href": re.compile('_.*.xls')}): jpx_url = conf.get_config('URL', 'jpx_base_url') url = jpx_url + href.get('href') save_path = r'D:\workspace\AnnounceFinancial\' + re.search(r'([^/]+?)?$', url).group() # ③xlsをローカルにダウンロードする download_file = html_util.download_file(url, save_path) # ④ダウンロードしたxlsをpandasで読み込みオブジェクト化する df = pd.read_excel(download_file[0], skiprows=2, dtype='str') # ⑤DBに入れるためにデータを色々整える # 列削除 df = df.drop(columns=['会社名', '決算期末', '市場区分']) # 列リネーム df = df.rename(columns={'発表予定日': 'ANNOUNCE_DATE', 'コード': 'STOCK_CODE', '業種名': 'INDUSTRY', '種別': 'FINANCIAL_TYPE'}) # 列追加 df = df.assign(INPUT_DATE=today) df = df.assign(ANNOUNCE_TIME='-') # 列の入れ替え df = df.loc[:, ['STOCK_CODE', 'ANNOUNCE_DATE', 'ANNOUNCE_TIME', 'FINANCIAL_TYPE', 'INDUSTRY', 'INPUT_DATE']] # 抽出 df = df.loc[(df['INDUSTRY'] != 'REIT') & (df['ANNOUNCE_DATE'] != '未定') & (df['STOCK_CODE'] != 'nan')] # データ置換 df['FINANCIAL_TYPE'] = df['FINANCIAL_TYPE'].replace({'第１四半期': '1Q', '第２四半期': '2Q', '第３四半期': '3Q', '本決算': ' 本'}) # 文字列追加 df['STOCK_CODE'] = df['STOCK_CODE'] + '-T' # 文字列 substring df['ANNOUNCE_DATE'] = df['ANNOUNCE_DATE'].str[:10] # ⑥整え終わったらDataFrameのリストに追加 df_list.append(df) # List内のDataFrameを合体 # ⑦ループ終了後、DataFrameのリストを一つのDataFrameに合体 df = pd.concat(df_list) # List＜Dictionary＞に変換 # ⑧一つのDataFrameをDictionary Listに変換する insert_list = df.to_dict(orient='records') # ⑨データベースに保存する dao.insert_announce_financial(insert_list)# urlにアクセスし、そのページをBeautifulSoupの型で返しますdef request_beautiful_soup(url): result = requests.get(url) content_type = result.encoding if result.encoding != 'ISO-8859-1' else None return BeautifulSoup(result.content, 'html.parser', from_encoding=content_type)# urlのファイルをダウンロードしますdef download_file(url, save_path): if not url or not save_path: return None try: result = urllib.request.urlretrieve(url, save_path) except urllib.request.ContentTooShortError: result = None return result# 決算発表スケジュールテーブルにMargeしますdef insert_announce_financial(insert_list): conn = get_connection() for announce_financial in insert_list: conn.execute("INSERT OR REPLACE INTO MST_ANNOUNCE_FINANCIAL " "VALUES(:STOCK_CODE, :ANNOUNCE_DATE, :ANNOUNCE_TIME, :FINANCIAL_TYPE, " ":INDUSTRY, :INPUT_DATE)", announce_financial) conn.commit()※投資はくれぐれも１００％ご自身の判断と責任の元で行ってください。↓↓押していただけると励みになります！
出産・子育て
ファッション
美容・コスメ
健康・ダイエット
生活・インテリア
料理・食べ物
ドリンク・お酒
ペット
趣味・ゲーム
映画・ＴＶ
音楽
読書・コミック
旅行・海外情報
園芸
スポーツ
アウトドア・釣り
車・バイク
パソコン・家電
そのほか
すべてのジャンル
