nikkei225jp.com data source
Usage
from pykabu.sources import nikkei225
Get economic schedule
schedule = nikkei225.get_schedule()
today = nikkei225.get_today_schedule()
Get market indices
indices = nikkei225.get_indices()
IndexItem
dataclass
A market index item
Source code in src/pykabu/sources/nikkei225.py
| @dataclass
class IndexItem:
"""A market index item"""
name: str
value: str
change: str
percent: str
|
RankItem
dataclass
A Nikkei 225 contribution ranking item
Source code in src/pykabu/sources/nikkei225.py
| @dataclass
class RankItem:
"""A Nikkei 225 contribution ranking item"""
name: str # 銘柄名
contribution: str # 寄与度
price: str # 現在値
change: str # 前日比
|
ScheduleItem
dataclass
A single schedule item from the economic calendar
Source code in src/pykabu/sources/nikkei225.py
| @dataclass
class ScheduleItem:
"""A single schedule item from the economic calendar"""
date_str: str
time: str
importance: str
indicator: str
result: str
forecast: str
previous: str
@property
def star_count(self) -> int:
"""Count the number of stars in importance"""
return self.importance.count("★")
|
star_count
property
Count the number of stars in importance
SectorRankItem
dataclass
A sector ranking item
Source code in src/pykabu/sources/nikkei225.py
| @dataclass
class SectorRankItem:
"""A sector ranking item"""
name: str # 業種名
change: str # 変動率 (e.g., "▲1.90%", "▼4.42%")
|
filter_schedule_by_importance(items, min_stars)
Filter schedule items by minimum importance (star count).
Source code in src/pykabu/sources/nikkei225.py
| def filter_schedule_by_importance(items: list[ScheduleItem], min_stars: int) -> list[ScheduleItem]:
"""Filter schedule items by minimum importance (star count)."""
return [item for item in items if item.star_count >= min_stars]
|
get_all_known_indices()
Return all known index codes.
Source code in src/pykabu/sources/nikkei225.py
| def get_all_known_indices() -> dict[str, str]:
"""Return all known index codes."""
return ALL_KNOWN_INDICES.copy()
|
get_default_indices()
Return the default index codes.
Source code in src/pykabu/sources/nikkei225.py
| def get_default_indices() -> dict[str, str]:
"""Return the default index codes."""
return DEFAULT_INDEX_CODES.copy()
|
get_indices(codes=None)
Fetch market index data from nikkei225jp.com (requires playwright).
Parameters:
| Name |
Type |
Description |
Default |
codes
|
dict[str, str] | None
|
Optional dict of code->name mappings. If None, uses DEFAULT_INDEX_CODES.
|
None
|
Source code in src/pykabu/sources/nikkei225.py
| def get_indices(codes: dict[str, str] | None = None) -> list[IndexItem]:
"""Fetch market index data from nikkei225jp.com (requires playwright).
Args:
codes: Optional dict of code->name mappings. If None, uses DEFAULT_INDEX_CODES.
"""
from playwright.sync_api import sync_playwright
if codes is None:
codes = DEFAULT_INDEX_CODES
items = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(BASE_URL, wait_until="domcontentloaded")
try:
page.wait_for_function(
"document.querySelector('#V511')?.textContent?.trim().length > 0",
timeout=500
)
except Exception:
pass
for code, name in codes.items():
try:
value = page.locator(f"#V{code}").text_content(timeout=500) or "-"
change = page.locator(f"#Z{code}").text_content(timeout=500) or "-"
percent = page.locator(f"#P{code}").text_content(timeout=500) or "-"
items.append(IndexItem(
name=name,
value=value.strip(),
change=change.strip(),
percent=percent.strip(),
))
except Exception:
pass
browser.close()
return items
|
get_rank225()
Fetch Nikkei 225 contribution ranking (requires playwright).
Returns:
| Type |
Description |
tuple[list[RankItem], list[RankItem]]
|
Tuple of (top_contributors, bottom_contributors)
|
Source code in src/pykabu/sources/nikkei225.py
| def get_rank225() -> tuple[list[RankItem], list[RankItem]]:
"""Fetch Nikkei 225 contribution ranking (requires playwright).
Returns:
Tuple of (top_contributors, bottom_contributors)
"""
from playwright.sync_api import sync_playwright
top_items: list[RankItem] = []
bottom_items: list[RankItem] = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(f"{BASE_URL}/chart/nikkei.php", wait_until="domcontentloaded")
# Wait for the ranking table to load (JS-rendered)
try:
page.wait_for_selector("#nkrnk .kiyoBdTD", timeout=5000)
except Exception:
pass
# Each row has 2 cells: left (top contributor) and right (bottom contributor)
# Each cell contains: .kiyoTDsp0 (name), .kiyoTDsp1 (contribution),
# .kiyoTDsp2 (price), .kiyoTDsp3 (change)
rows = page.query_selector_all("#nkrnk tr.kiyoBdTR")
for row in rows:
cells = row.query_selector_all("td.kiyoBdTD")
if len(cells) >= 2:
# Left cell: top contributor (寄与度上位)
top_item = _extract_cell_data(cells[0])
if top_item:
top_items.append(top_item)
# Right cell: bottom contributor (寄与度下位)
bottom_item = _extract_cell_data(cells[1])
if bottom_item:
bottom_items.append(bottom_item)
browser.close()
return top_items, bottom_items
|
get_schedule()
Fetch all schedule items from nikkei225jp.com/schedule/
Source code in src/pykabu/sources/nikkei225.py
| def get_schedule() -> list[ScheduleItem]:
"""Fetch all schedule items from nikkei225jp.com/schedule/"""
html = fetch_page(BASE_URL, "/schedule/")
parser = _ScheduleTableParser()
parser.feed(html)
return parser.items
|
get_schedule_for_date(target)
Get schedule items for a specific date.
Source code in src/pykabu/sources/nikkei225.py
| def get_schedule_for_date(target: date) -> list[ScheduleItem]:
"""Get schedule items for a specific date."""
return [item for item in get_schedule() if _matches_date(item, target)]
|
get_sector_rank()
Fetch sector ranking (requires playwright).
Returns:
Source code in src/pykabu/sources/nikkei225.py
| def get_sector_rank() -> tuple[list[SectorRankItem], list[SectorRankItem]]:
"""Fetch sector ranking (requires playwright).
Returns:
Tuple of (top_gainers, top_losers)
"""
from playwright.sync_api import sync_playwright
gainers: list[SectorRankItem] = []
losers: list[SectorRankItem] = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(f"{BASE_URL}/chart/gyoushu.php", wait_until="domcontentloaded")
# Wait for the ranking table to load (JS-rendered)
try:
page.wait_for_selector("#gyornk .tdG", timeout=5000)
except Exception:
pass
# Two nested tables inside #gyornk: left (gainers), right (losers)
tables = page.query_selector_all("#gyornk td.tptd > table")
if len(tables) >= 2:
# Left table: top gainers (値上がり率 TOP10)
gainer_rows = tables[0].query_selector_all("tr.trG")
for row in gainer_rows:
cell = row.query_selector("td.tdG")
if cell:
per_el = cell.query_selector(".perG")
name_el = cell.query_selector(".texG")
if per_el and name_el:
gainers.append(SectorRankItem(
name=(name_el.text_content() or "").strip(),
change=(per_el.text_content() or "").strip(),
))
# Right table: top losers (値下がり率 TOP10)
loser_rows = tables[1].query_selector_all("tr.trG")
for row in loser_rows:
cell = row.query_selector("td.tdG")
if cell:
per_el = cell.query_selector(".perG")
name_el = cell.query_selector(".texG")
if per_el and name_el:
losers.append(SectorRankItem(
name=(name_el.text_content() or "").strip(),
change=(per_el.text_content() or "").strip(),
))
browser.close()
return gainers, losers
|
get_today_schedule()
Get schedule items for today.
Source code in src/pykabu/sources/nikkei225.py
| def get_today_schedule() -> list[ScheduleItem]:
"""Get schedule items for today."""
return get_schedule_for_date(date.today())
|
get_tomorrow_schedule()
Get schedule items for tomorrow.
Source code in src/pykabu/sources/nikkei225.py
| def get_tomorrow_schedule() -> list[ScheduleItem]:
"""Get schedule items for tomorrow."""
return get_schedule_for_date(date.today() + timedelta(days=1))
|
get_week_schedule()
Get schedule items for this week (today through 7 days).
Source code in src/pykabu/sources/nikkei225.py
| def get_week_schedule() -> list[ScheduleItem]:
"""Get schedule items for this week (today through 7 days)."""
all_items = get_schedule()
today = date.today()
week_dates = [today + timedelta(days=i) for i in range(7)]
return [item for item in all_items if any(_matches_date(item, d) for d in week_dates)]
|