導入
NFC Market Liveは、24時間365日、人間の代わりに相場を監視し続けるAI経済ニュース局です。 前回の記事では、システム全体を統括する「司令塔（main_station.py）」について解説しました。
しかし、ただ漫然とループを回すだけでは「放送局」とは呼べません。 「いつ、どのタイミングで、どの指標を発表するか？」 投資家にとって最も価値があるのは「情報の鮮度とタイミング」です。
今回は、NFCの心臓部とも言える「編成局（nfc_scheduler.py）」のコードと、その泥臭い仕組みを公開します。
編成局の2つの役割：「定時枠」と「特番割り込み」
実際のテレビ局の編成をイメージしてみてください。基本的には決まった時間割（定時番組）が流れていますが、大事件が起きれば「速報（特番）」が割り込みますよね。
NFCのスケジューラーも、全く同じ思想で設計しています。
1. 60分サイクルの「定時ローテーション」
毎時0分〜60分の間に、以下のようなゾーン制を敷いています。
- 00分〜20分: 日本市場ゾーン（日銀レート、国内CPIなど）
- 20分〜40分: 米国市場ゾーン（米国債利回り、雇用統計など）
- 40分〜: エネルギー動向やAIによる経済ラジオ（休憩枠）
2. 重要指標発表時の「特番（速報）割り込み」
これがNFC最大の武器です。例えば「米・雇用統計」が21:30に発表される日。スケジューラーはそれを事前に察知し、発表1分前から発表後8分間は、すべての定時番組をストップして「雇用統計の速報システム」に画面をジャックさせます。
実際の制御コード（一部抜粋）はこちらです。
def get_current_program(self) -> Tuple[str, str]:
now = datetime.datetime.now()
minute = now.minute
# 1. 速報・特番割り込みチェック (Highest Priority)
for event in self.schedule_db:
if event['date'] != now.date(): continue
try:
h, m = map(int, event["time"].split(":"))
target_dt = now.replace(hour=h, minute=m, second=0, microsecond=0)
except: continue
# 特番の優先時間（開始1分前 〜 開始後8分間）
start_dt = target_dt - datetime.timedelta(minutes=1)
end_dt = target_dt + datetime.timedelta(minutes=8)
if start_dt <= now <= end_dt:
logger.info(f"⚡️ [Scheduler] 速報割り込み: {event['label']}")
return event["producer"], event["label"], True
# 2. 定時放送ローテーション (特番がない場合は通常営業)
selected = None
if 0 <= minute < 20:
zone_list = self.rotation_jp
slot_duration = 3.3
base_idx = int(minute / slot_duration) % len(zone_list)
selected = zone_list[base_idx]
# ...(中略：米国・エネルギーなどのローテーション処理)
return selected[0], selected[1], False世界中から「今日の時刻表」をかき集める泥臭いスクレイピング
では、スケジューラーはどうやって「今日の特番予定（経済指標の発表時刻）」を知るのでしょうか？
実は、毎日システムが起動した瞬間に、世界中の政府機関や中央銀行のサイトへ自動でアクセスし、その日の発表スケジュールをかき集めています。
- 米国の指標: セントルイス連銀のAPI（FRED API）からJSONで取得（※APIキーは環境変数で保護）。
- 日本の指標: 財務省、内閣府、日銀のWebサイトから、HTMLのテーブルやXMLを力技でスクレイピング。
以下は、財務省のサイトから「国債入札（JGB）」の予定を引っこ抜くコードです。
def _fetch_mof_jgb_schedule(self) -> List[Dict]:
base_url = "https://www.mof.go.jp/jgbs/auction/calendar/{}.htm"
events = []
now = datetime.datetime.now()
for i in range(2):
target_month = now + datetime.timedelta(days=30 * i)
yyyymm = target_month.strftime("%y%m")
y_val = target_month.year
url = base_url.format(yyyymm)
try:
res = requests.get(url, timeout=5)
if res.status_code == 200:
res.encoding = 'utf-8'
dfs = pd.read_html(io.StringIO(res.text)) # pandasでHTMLテーブルを解析
for df in dfs:
if "入札" not in str(df): continue
for _, row in df.iterrows():
# テーブル内のテキストを結合して正規表現で日付を抽出
row_text = " ".join([str(x) for x in row.values if pd.notna(x)])
match = re.search(r'(\d{1,2})\s*月\s*(\d{1,2})\s*日', row_text)
if match:
m, d = map(int, match.groups())
label = None
# 割引、利付などのキーワードで国債の種類を特定
for cell in row.values:
s = str(cell).strip()
if any(k in s for k in ["利付", "割引", "国庫券", "流動性", "物価連動"]):
label = s; break
if label:
events.append({"date": datetime.date(y_val, m, d), "time": "12:35", "producer": "jgb_auction", "label": f"入札結果: {label}", "is_fixed": False})
except: continue
return eventsFREDのような綺麗なAPIなら簡単なのですが、日本のお役所のサイトはExcelをそのままHTMLにしたような構造が多く、BeautifulSoupやpandasを使ってテキストを正規表現でこじ開けるという、非常に泥臭い処理を行っています。
非エンジニアが「落ちないシステム」を作るために
このスケジューラーは、外部サイトに依存しています。もし日銀のサイトがメンテナンス中でスクレイピングに失敗したら？
システム全体が巻き込まれてクラッシュしては、放送局として失格です。 そのため、各スクレイピング関数の随所に try-except を仕込み、「予定が取得できなかったら、とりあえず無視して通常放送を続ける」という安全側に倒した設計（フェイルセーフ）にしています。
try:
res = requests.get(url, timeout=5) # 5秒で応答がなければ諦める
# ...解析処理...
except Exception as e:
logger.error(f"GDP Schedule Parse Error: {e}")
pass # エラーを吐き出しつつ、処理は止めない次回予告：素材を動画に仕立て上げる「編集室」
今回はシステムのリズムを作る「編成局（スケジューラー）」について解説しました。 各プロデューサーがデータを集めてきても、そのままでは放送できません。
次回は、集まった画像やテキスト素材を所定のレイアウトに美しく配置し、一本の動画ファイルへとレンダリングする「編集室（nfc_video_composer.py）」の役割と、美しいUIへのこだわりに迫ります。
（※現在、裏ではこのシステム構築ノウハウを活かし、AWSの東京リージョンでミリ秒のレイテンシを削りながら動く「仮想通貨アービトラージbot（Rust）」のプロジェクトも密かに進行中です。こちらもいずれ公開予定ですのでお楽しみに）
コメント