Der Titel ist scheiße lang. .. 9uant betreibt quantitative Analyse und Datenwissenschaft im Finanzbereich. Wir machen auch Twitter, also wenn Sie interessiert sind, folgen Sie uns bitte!
Wie aus dem Titel hervorgeht, werden wir den Code freigeben, um die XBRL des Jahresabschlusses schnell herunterzuladen. Ich möchte einen Kommentar nach dem anderen schreiben. Führen Sie die folgenden zwei Schritte aus.
--DataFrame den Link zu XBRL
import os
import glob
import shutil
import re
import time
from datetime import date, timedelta, datetime
from dateutil.relativedelta import relativedelta
import requests
from bs4 import BeautifulSoup
import urllib3
from urllib3.exceptions import InsecureRequestWarning
urllib3.disable_warnings(InsecureRequestWarning)
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import Select
from selenium.webdriver.common.by import By
import zipfile
import numpy as np
import pandas as pd
import json
EDINET Einfach, weil EDINET eine API hat.
def edinet_xbrl_link(annual=True, quarter=True, codes=None, year=0,month=0,day=0):
'''
Erstellen Sie einen DataFrame für XBRL-Links für bestimmte Unternehmen oder alle Wertpapiere und Quartalsberichte
Parameters:
annual: bool, default True
Holen Sie sich einen Wertpapierbericht, wenn True
quarter: bool, default Ture
Holen Sie sich einen vierteljährlichen Bericht, wenn True
codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
Holen Sie sich Daten für alle Unternehmen, wenn keine
Wenn Sie einen Aktiencode angeben, werden nur die Daten dieser Unternehmen erfasst.:point_up_tone4:
year, month, day: int, default 0
Geben Sie an, vor wie vielen Tagen die Daten abgerufen werden sollen(Bis zu 5 Jahre)
Returns:
database: pandas.DataFrame
database['code']: str
5-stelliger Wertpapiercode
database['type']: str
'annual' or 'quarter'
database['date']: datetime.date
Veröffentlichungsdatum
database['title']: str
Titel
database['URL']: str
URL zum Herunterladen der XBRL-Zip-Datei
'''
edinet_url = "https://disclosure.edinet-fsa.go.jp/api/v1/documents.json"
#Vereinheitlichen Sie Codes zu einem Zeichentyp-Array.
if codes != None:
if type(codes) in (str, int, float):
codes = [int(codes)]
for code in codes:
#Konvertieren Sie den 4-stelligen Wertpapiercode in einen 5-stelligen
if len(str(int(code)))==4:
code = str(int(code))+'0'
#Drehen Sie die for-Anweisung mit dem Datum / Uhrzeit-Typ
def date_range(start, stop, step = timedelta(1)):
current = start
while current < stop:
yield current
current += step
#Bereiten Sie einen DataFrame vor, um das Ergebnis zu speichern
database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
for d in date_range(date.today()-relativedelta(years=year, months=month, days=day)+relativedelta(days=1), date.today()+relativedelta(days=1)):
#Greifen Sie auf die EDINET-API zu
d_str = d.strftime('%Y-%m-%d')
params = {'date' : d_str, 'type' : 2}
res = requests.get(edinet_url, params=params, verify=False)
json_res = json.loads(res.text)
time.sleep(5)
#Wenn Sie nicht normal zugreifen können
if json_res['metadata']['status']!='200':
print(d_str, 'not accessible')
continue
print(d_str, json_res['metadata']['resultset']['count'])#Datum und Nummer anzeigen
#Im Falle von 0
if len(json_res['results'])==0:
continue
df = pd.DataFrame(json_res['results'])[['docID', 'secCode', 'ordinanceCode', 'formCode','docDescription']]
df.dropna(subset=['docID'], inplace=True)
df.dropna(subset=['secCode'], inplace=True)
df.rename(columns={'secCode': 'code', 'docDescription': 'title'}, inplace=True)
df['date'] = d
df['URL'] = df['docID']
df['URL'] = "https://disclosure.edinet-fsa.go.jp/api/v1/documents/" + df['URL']
#Extrahieren Sie nur den angegebenen Wertpapiercode
if codes != None:
df = df[df['code'] in codes]
if annual == True:
df1 = df[(df['ordinanceCode']=='010') & (df['formCode']=='030000')]
df1['type'] = 'annual'
database = pd.concat([database, df1[['code', 'type', 'date','title', 'URL']]], axis=0, join='outer').reset_index(drop=True)
if quarter == True:
df2 = df[(df['ordinanceCode']=='010') & (df['formCode']=='043000')]
df2['type'] = 'quarter'
database = pd.concat([database, df2[['code', 'type', 'date','title', 'URL']]], axis=0, join='outer').reset_index(drop=True)
return database
TDNET Verwenden Sie "Selen", um Daten von TDNET zu sammeln. Da nur 200 freie Wortsuchergebnisse angezeigt werden, haben wir eine Funktion zum Suchen nach Wertpapiercode und eine Funktion zum separaten Suchen nach Datum erstellt.
def tdnet_xbrl_link_by_code(codes):
'''
Erstellen Sie einen DataFrame, der die Abschlüsse des angegebenen Unternehmens mit XBRL verknüpft
Parameters:
codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
Holen Sie sich Daten für alle Unternehmen, wenn keine
Returns:
database: pandas.DataFrame
database['code']: str
5-stelliger Wertpapiercode
database['type']: str
'annual' or 'quarter'
database['date']: datetime.date
Veröffentlichungsdatum
database['title']: str
Titel
database['URL']: str
URL zum Herunterladen der XBRL-Zip-Datei
'''
#Vereinheitlichen Sie Codes zu einem Zeichentyp-Array.
if type(codes) in (str, int, float):
codes = [int(codes)]
for i, code in enumerate(codes):
#Konvertieren Sie den 4-stelligen Wertpapiercode in einen 5-stelligen
if len(str(int(code)))==4:
codes[i] = str(int(code))+'0'
database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
for code in codes:
#Starten Sie den Browser
chromeOptions = webdriver.ChromeOptions()
chromeOptions.add_argument('--headless') #Browser ausblenden
driver = webdriver.Chrome(options=chromeOptions)
driver.get("https://www.release.tdnet.info/onsf/TDJFSearch/I_head")
#Suchwort senden
duration = driver.find_element_by_name('t0')
select = Select(duration)
select.options[-1].click()
inputElement = driver.find_element_by_id("freewordtxt")
inputElement.send_keys(code)
inputElement.send_keys(Keys.RETURN)
time.sleep(5)
#Gehen Sie zu dem Rahmen, in dem die Suchergebnisse angezeigt werden
iframe = driver.find_element_by_name("mainlist")
driver.switch_to.frame(iframe)
#Beendet die Verarbeitung, wenn 0 Suchergebnisse vorhanden sind
if driver.find_element_by_id("contentwrapper").text == 'Die relevanten Informationen zur rechtzeitigen Offenlegung wurden nicht gefunden.':
return database
#Lesen Sie die Daten aus jeder Zeile in der Suchergebnistabelle
table = driver.find_element_by_id("maintable")
trs = table.find_elements(By.TAG_NAME, "tr")
for i in range(len(trs)):
title = trs[i].find_elements(By.TAG_NAME, "td")[3].text
#Wählen Sie den Jahresabschluss des angegebenen Unternehmens aus, in dem XBRL vorhanden ist, nicht die Korrekturdokumente
if ('Jahresabschluss' in title) and ('Korrektur' not in title) and (len(trs[i].find_elements(By.TAG_NAME, "td")[4].text)!=0) and (code==trs[i].find_elements(By.TAG_NAME, "td")[1].text):
date = trs[i].find_elements(By.TAG_NAME, "td")[0].text[:10]
date = datetime.strptime(date, '%Y/%m/%d').date()
url = trs[i].find_elements(By.TAG_NAME, "td")[4].find_element_by_tag_name("a").get_attribute("href")
database = database.append(pd.Series([code,'brief',date,title,url], index=database.columns), ignore_index=True)
driver.quit()
return database
def tdnet_xbrl_link_by_date(date=None):
'''
Erstellen Sie einen DataFrame, der das angegebene Datum oder alle Abschlüsse mit XBRL verknüpft
Parameters:
date: None or str ('yyyy/mm/dd'), default None
Daten für alle Daten abrufen, wenn Keine
Returns:
database: pandas.DataFrame
database['code']: str
5-stelliger Wertpapiercode
database['type']: str
'annual' or 'quarter'
database['date']: datetime.date
Veröffentlichungsdatum
database['title']: str
Titel
database['URL']: str
URL zum Herunterladen der XBRL-Zip-Datei
'''
database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
#Starten Sie den Browser
chromeOptions = webdriver.ChromeOptions()
chromeOptions.add_argument('--headless') #Browser ausblenden
driver = webdriver.Chrome(options=chromeOptions)
driver.get("https://www.release.tdnet.info/inbs/I_main_00.html")
duration = driver.find_element_by_name('daylist')
select = Select(duration)
for i in range(1, len(select.options)):
driver.get("https://www.release.tdnet.info/inbs/I_main_00.html")
duration = driver.find_element_by_name('daylist')
select = Select(duration)
d = datetime.strptime(select.options[i].text[:10], '%Y/%m/%d').date()
print(select.options[i].text)
if (date == None) or (date == select.options[i].text[:10]):
select.options[i].click()
time.sleep(5)
#Gehen Sie zu dem Rahmen, in dem die Suchergebnisse angezeigt werden
iframe = driver.find_element_by_id("main_list")
driver.switch_to.frame(iframe)
#Beendet die Verarbeitung, wenn 0 Suchergebnisse vorhanden sind
if driver.find_element_by_id("kaiji-text-1").text!='Informationen offengelegt in':
continue
#Setzen Sie die Verarbeitung bis zur letzten Seite fort
while True:
#Lesen Sie die Daten aus jeder Zeile in der Suchergebnistabelle
table = driver.find_element_by_id("main-list-table")
trs = table.find_elements(By.TAG_NAME, "tr")
for i in range(len(trs)):
title = trs[i].find_elements(By.TAG_NAME, "td")[3].text
#Wählen Sie den Jahresabschluss des angegebenen Unternehmens aus, in dem XBRL vorhanden ist, nicht die Korrekturdokumente
if ('Jahresabschluss' in title) and ('Korrektur' not in title) and (len(trs[i].find_elements(By.TAG_NAME, "td")[4].text)!=0):
code = trs[i].find_elements(By.TAG_NAME, "td")[1].text
url = trs[i].find_elements(By.TAG_NAME, "td")[4].find_element_by_tag_name("a").get_attribute("href")
database = database.append(pd.Series([code, 'brief', d, title,url], index=database.columns), ignore_index=True)
if len(driver.find_element_by_class_name("pager-R").text)!=0:
driver.find_element_by_class_name("pager-R").click()
time.sleep(5)
else:
#Beendet die Verarbeitung, wenn das Wort "Weiter" nicht vorhanden ist
break
driver.quit()
return database
def dl_xbrl_zip(codes=None, database):
'''
Laden Sie die XBRL-Zip-Datei herunter, indem Sie auf den DataFrame verweisen, in dem die Links zur XBRL aufgeführt sind.
Parameters:
codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
Holen Sie sich XBRL für alle Unternehmen, wenn keine
database: pandas.DataFrame
database['code']: str
5-stelliger Wertpapiercode
database['type']: str
'annual' or 'quarter'
database['date']: datetime.date
Veröffentlichungsdatum
database['title']: str
Titel
database['URL']: str
URL zum Herunterladen der XBRL-Zip-Datei
Returns:
None
'''
database.dropna(subset=['code'], inplace=True)
database = database.reset_index(drop=True)
#Vereinheitlichen Sie Codes zu einem Zeichenarray
if codes == None:
codes = [None]
else:
if type(codes) in (str, int, float):
codes = [int(codes)]
for i, code in enumerate(codes):
#Konvertieren Sie den 4-stelligen Wertpapiercode in einen 5-stelligen
if len(str(int(code)))==4:
codes[i] = str(int(code))+'0'
for code in codes:
if code == None:
df_company = database
else:
df_company = database[database['code']==code]
df_company = df_company.reset_index(drop=True)
#Verwenden Sie den Wertpapiercode als Verzeichnisnamen
dir_path = database.loc[i,'code']
if os.path.exists(dir_path)==False:
os.mkdir(dir_path)
#Laden Sie die Zip-Datei aus jeder Zeile der extrahierten Liste vom Link zu XBRL herunter
for i in range(df_company.shape[0]):
#Beim Zugriff auf EDINET
if (df_company.loc[i,'type'] == 'annual') or (df_company.loc[i,'type'] == 'quarter'):
params = {"type": 1}
res = requests.get(df_company.loc[i,'URL'], params=params, stream=True)
if df_company.loc[i,'type'] == 'annual':
#Der Dateiname des Wertpapierberichts lautet"yyyy_0.zip"
filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r"_0.zip"
elif df_company.loc[i,'type'] == 'quarter':
if re.search('Zeitraum', df_company.loc[i,'title']) == None:
#Der Dateiname des Quartalsberichts unbekannter Periode lautet"yyyy_unknown_docID.zip"
filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r'_unknown_' + df_company.loc[i,'URL'][-8:] + r'.zip'
else:
#Der Dateiname des Quartalsberichts lautet"yyyy_quarter.zip"
filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r'_' + df_company.loc[i,'title'][re.search('Zeitraum', df_company.loc[i,'title']).end()] + r'.zip'
#Beim Zugriff auf TDNET
elif df_company.loc[i,'type'] == 'brief':
res = requests.get(df_company.loc[i,'URL'], stream=True)
#Füllen Sie die leeren Zeichen aus
s_list = df_company.loc[i,'title'].split()
s = ''
for i in s_list:
s += i
filename = df_company.loc[i,'date'][:4] + r'_' + s[re.search('Zeitraum', s).end()] + r'_brief.zip'
#Wenn eine gleichnamige Zip-Datei vorhanden ist, wird diese nicht überschrieben.
if os.path.exists(filename):
print(df_company.loc[i,'code'],df_company.loc[i,'date'],'already exists')
continue
#Laden Sie die Zip-Datei nur herunter, wenn Sie normal darauf zugreifen können
if res.status_code == 200:
with open(filename, 'wb') as file:
for chunk in res.iter_content(chunk_size=1024):
file.write(chunk)
print(df_company.loc[i,'code'],df_company.loc[i,'date'],'saved')
print('done!')
return None
Recommended Posts