投資判断にデータ分析を活用する——弊社の情報収集フロー

弊社のデータ分析活用の背景

株式会社WRはEC事業・システム開発を軸に事業を展開しており、日常業務の中でデータ収集と分析が意思決定の根幹を担っています。本記事では弊社が実際に使っている情報収集・分析フローを公開します。

収集する情報の種類

弊社が継続的に収集・分析しているデータは主に以下の4カテゴリです。

情報収集システムの構成

[データソース]
├── Amazon Product API / スクレイピング
├── 楽天商品検索API
├── Qoo10スクレイピング（Puppeteer）
└── 為替API（Open Exchange Rates）

[収集・処理]
├── Node.js cronジョブ（1時間ごと）
│   └── 価格データをPostgreSQLに保存
├── GAS（1日1回）
│   └── 各モールランキングを収集
│   └── スプレッドシートに書き込み
└── Python バッチ（日次）
    └── データ集計・分析レポート生成

[通知・共有]
├── Slack Webhook（価格アラート・異常検知）
└── Google スプレッドシート（日次ダッシュボード）

価格データの保存設計

-- PostgreSQL スキーマ
CREATE TABLE price_history (
    id            BIGSERIAL PRIMARY KEY,
    product_code  VARCHAR(50)  NOT NULL,
    platform      VARCHAR(20)  NOT NULL,  -- amazon, rakuten, qoo10
    price         INTEGER      NOT NULL,
    stock_status  VARCHAR(20)  DEFAULT 'in_stock',
    rank          INTEGER,                -- ランキング順位
    collected_at  TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    created_at    TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

-- 直近30日分だけ高頻度で分析するインデックス
CREATE INDEX idx_price_history_product_platform_date
    ON price_history (product_code, platform, collected_at DESC);

Pythonでの価格トレンド分析

import pandas as pd
import numpy as np
from sqlalchemy import create_engine

engine = create_engine(os.environ['DATABASE_URL'])

def analyze_price_trend(product_code: str, days: int = 30) -> dict:
    """直近N日間の価格トレンドを分析"""
    query = """
        SELECT
            platform,
            collected_at::date AS date,
            AVG(price) AS avg_price,
            MIN(price) AS min_price
        FROM price_history
        WHERE product_code = %(code)s
          AND collected_at >= NOW() - INTERVAL '%(days)s days'
        GROUP BY platform, collected_at::date
        ORDER BY platform, date
    """
    df = pd.read_sql(query, engine, params={'code': product_code, 'days': days})

    results = {}
    for platform, group in df.groupby('platform'):
        prices = group['avg_price'].values

        # 線形回帰でトレンドを計算
        x = np.arange(len(prices))
        slope, intercept = np.polyfit(x, prices, 1)

        results[platform] = {
            'current_price': int(prices[-1]),
            'avg_price':     int(prices.mean()),
            'min_price':     int(prices.min()),
            'max_price':     int(prices.max()),
            'trend_slope':   round(slope, 2),          # 正：上昇、負：下降
            'trend_direction': '上昇' if slope > 5 else ('下降' if slope < -5 else '横ばい'),
            'price_range':   int(prices.max() - prices.min()),
        }

    return results

仕入れ判断ロジック

def evaluate_purchase_opportunity(
    product_code: str,
    purchase_cost: float,
    target_margin: float = 0.20,
) -> dict:
    """仕入れ判断を行う"""

    trend = analyze_price_trend(product_code)

    # 最も安い販売価格プラットフォームを確認
    best_platform = min(trend.items(), key=lambda x: x[1]['current_price'])
    platform_name, data = best_platform

    selling_price = data['current_price']
    margin = (selling_price - purchase_cost) / selling_price

    # 判断ロジック
    is_good_opportunity = all([
        margin >= target_margin,
        data['trend_direction'] in ['横ばい', '上昇'],  # 価格下落中は見送り
        data['current_price'] <= data['avg_price'] * 1.1,  # 平均の110%以内
    ])

    return {
        'product_code':    product_code,
        'platform':        platform_name,
        'selling_price':   selling_price,
        'purchase_cost':   purchase_cost,
        'margin_pct':      round(margin * 100, 1),
        'trend':           data['trend_direction'],
        'recommendation':  '✅ 仕入れ推奨' if is_good_opportunity else '❌ 見送り',
        'reason':          _build_reason(margin, target_margin, data),
    }

def _build_reason(margin, target, data):
    reasons = []
    if margin < target:
        reasons.append(f"利益率{margin*100:.1f}%（目標{target*100:.0f}%未満）")
    if data['trend_direction'] == '下降':
        reasons.append("価格が下降トレンド")
    return '、'.join(reasons) or '全条件クリア'

Slackへの定期配信

import requests

def send_daily_summary():
    """毎朝の仕入れ候補サマリーをSlackに送る"""
    opportunities = []
    for product in get_watchlist_products():
        result = evaluate_purchase_opportunity(
            product['code'],
            product['purchase_cost'],
        )
        if result['recommendation'] == '✅ 仕入れ推奨':
            opportunities.append(result)

    if not opportunities:
        message = "本日の仕入れ推奨商品はありません"
    else:
        lines = [f"*📊 本日の仕入れ候補: {len(opportunities)}件*\n"]
        for o in opportunities[:5]:
            lines.append(
                f"• {o['product_code']}: ¥{o['selling_price']:,} "
                f"（利益率{o['margin_pct']}% / {o['trend']}）"
            )
        message = '\n'.join(lines)

    requests.post(
        os.environ['SLACK_WEBHOOK'],
        json={'text': message},
    )

まとめ

データ分析を意思決定に組み込むことで、経験と勘だけに頼らない根拠ある仕入れ判断が可能になります。弊社では価格収集・分析・通知・判断支援の一連のフローを自動化することで、業務効率と収益性の向上を実現しています。

データ分析システム・業務自動化ツールの開発についてはお気軽にご相談ください。