開発環境の準備

本プロジェクトは以下の環境で動作確認を行っています。

• OS: Windows, macOS, Linux対応
• Python 3.7以降
• 必要ライブラリ: pandas（データ処理）, requests（HTTP通信）, re（テキスト解析）, akshare（金融データ取得）, matplotlib（グラフ描画）, dataframe-image（表の画像化）

投資信託データの取得手法

金融データの取得は、専用ライブラリを利用する方法と、Web APIを直接叩く方法の2種類が考えられます。

Akshareを用いた履歴データの取得

Akshareを利用すれば、少ないコード量で履歴データを取得可能です。ただし、期間指定には直接対応していないため、全量取得後に日付でフィルタリングするアプローチをとります。

import akshare as ak

def fetch_historical_nav(symbol):
    nav_df = ak.fund_em_open_fund_info(fund=symbol, indicator='单位净值走势')
    return nav_df

Web APIを直接呼び出す手法

東方財富のAPIエンドポイントを利用することで、期間とページネーションを指定したデータ取得が可能になります。

import requests
import re
import pandas as pd

def fetch_page_content(symbol, date_from, date_to, page_num=1, per_page=40):
    endpoint = f"http://fund.eastmoney.com/f10/F10DataApi.aspx?type=lsjz&code={symbol}&page={page_num}&sdate={date_from}&edate={date_to}&per={per_page}"
    response = requests.get(endpoint)
    return response.text

def extract_dataframe(raw_html):
    match = re.search(r'content:"<table(.*)</table>",', raw_html)
    if match:
        table_segment = f"<table{match.group(1)}</table>"
        return pd.read_html(table_segment)[0]
    return pd.DataFrame()

def compile_fund_history(symbol, date_from, date_to):
    initial_html = fetch_page_content(symbol, date_from, date_to)
    total_pages_match = re.search(r'pages:(.*),', initial_html)
    if not total_pages_match:
        return "データ取得エラー"
    
    max_pages = int(total_pages_match.group(1))
    all_frames = []
    
    for current_page in range(max_pages):
        html_content = initial_html if current_page == 0 else fetch_page_content(symbol, date_from, date_to, page_num=current_page+1)
        df_chunk = extract_dataframe(html_content)
        all_frames.append(df_chunk)
        
    return pd.concat(all_frames, ignore_index=True)

今回は手軽さからAkshareを採用し、定期実行でローカルにCSVとしてデータを蓄積する設計とします。

データ蓄積と定期更新

毎日午前3時に監視銘柄のデータを更新し、ローカルに保存するジョブを定義します。

import time
import random
import akshare as ak

def refresh_surveillance_targets():
    try:
        with open('./config/monitored_funds.txt', 'r') as f:
            targets = [line.strip() for line in f.readlines()]
            
        for target_code in targets:
            history = ak.fund_em_open_fund_info(symbol=target_code, indicator='单位净值走势')
            sorted_history = history.sort_values(by='净值日期', ascending=False)
            sorted_history.to_csv(f"./data/fund_{target_code}.csv", index=False)
            time.sleep(random.randint(2, 6))
            
        return "データ更新完了"
    except Exception as err:
        return f"エラー発生: {err}"

def query_local_nav(target_code, start, end):
    df = pd.read_csv(f'./data/fund_{target_code}.csv')
    filtered = df.query(f"'{start}' <= 净值日期 <= '{end}'")
    return filtered

データの可視化と画像出力

取得したデータは、件数に応じて表形式またはトレンドグラフとして画像出力し、チャットボット経由で配信します。

表形式での画像出力

データ件数が30件以下の場合は、dataframe-imageを用いてDataframeをそのまま画像化します。最大値・最小値のハイライトやパーセント表示のスタイリングも施します。

import dataframe_image as dfi

def convert_table_to_image(df, code, start, end):
    if len(df) <= 1:
        dfi.export(df, f'./assets/{code}_{start}_{end}_table.png')
        return
        
    styled_df = df.style.highlight_max(subset=['单位净值'], color='salmon') \
                       .highlight_min(subset=['单位净值'], color='lightgreen') \
                       .format({'日增长率': '{:}%'})
    dfi.export(styled_df, f'./assets/{code}_{start}_{end}_table.png')

トレンドグラフの描画

データ件数が30件を超える場合は、matplotlibで折れ線グラフを描画し、ピークとボトムにアノテーションを付与します。

import matplotlib.pyplot as plt

def render_trend_chart(df, code, start, end):
    fig, ax = plt.subplots(figsize=(10, 5))
    ax.spines['right'].set_visible(False)
    ax.spines['top'].set_visible(False)
    ax.grid(axis='y', linestyle='--', alpha=0.7)
    
    peak = df.loc[df['单位净值'].idxmax()]
    trough = df.loc[df['单位净值'].idxmin()]
    
    ax.annotate(f'Max: {peak[1]}', xy=(peak[0], peak[1]), color='red')
    ax.annotate(f'Min: {trough[1]}', xy=(trough[0], trough[1]), color='green')
    
    ax.plot(df['净值日期'], df['单位净值'], marker='o', markersize=3, linestyle='-', color='teal')
    plt.title('投資信託基準価額トレンド')
    plt.xticks(rotation=45)
    plt.xlabel('日付')
    plt.ylabel('基準価額')
    plt.tight_layout()
    plt.savefig(f'./assets/{code}_{start}_{end}_chart.png')

出力の分岐処理

import os

def generate_response_visual(code, start, end):
    file_path = f'./assets/{code}_{start}_{end}_visual.png'
    if os.path.exists(file_path):
        return file_path
        
    dataset = query_local_nav(code, start, end)
    
    if len(dataset) <= 30:
        convert_table_to_image(dataset, code, start, end)
    else:
        sorted_df = dataset.sort_values(by='净值日期')
        render_trend_chart(sorted_df, code, start, end)
        
    return file_path

DingTalk連携と画像配信

DingTalkのBotは画像を直接送信できないため、Markdown形式で画像URLを送信する必要があります。Flaskの静的ファイルホスティング機能を利用して、生成した画像をHTTP経由で配信できるようにします。

from flask import Flask

app = Flask(__name__, static_folder='./assets', static_url_path='/images')

これにより、http://[サーバーIP]:[ポート]/images/[画像ファイル名]というURLで画像にアクセス可能になり、BotからのメッセージpayloadにMarkdownのimgタグとして埋め込むことで画像表示を実現します。

トラブルシューティング

Linux環境でのdataframe-imageエラー

Linux環境でdataframe-imageを実行すると、Chromeの起動に失敗することがあります。特にRootユーザーで実行している場合、--no-sandboxオプションが必須となりますが、ライブラリ側でこれがサポートされていない場合は専用の一般ユーザーを作成して実行する必要があります。

useradd botuser
chown -R botuser /path/to/project
su botuser

中国語フォントの文字化け

生成した画像やグラフで中国語が四角い枠（豆腐）になってしまう場合、サーバーに中国語フォントがインストールされていないことが原因です。以下の手順でフォントを追加し、キャッシュをクリアします。

mkdir -p /usr/share/fonts/custom_fonts
# ローカルからSimHei.ttfなどをアップロード
fc-cache -f -v

matplotlibを使用するスクリプト内でも、デフォルトフォントを変更する設定を忘れずに追加してください。