e-stat APIとかRESAS APIとか，Web APIは便利ですよね。
今回はPythonでAPIを作る方法をメモしておくことにします。
APIで提供する機能は「機械学習による予測」としておきます。
今回は
ところまでを書きます。デプロイについては今後勉強して書きます。
要件を決める
作るものをざっくり決めておきます。
土地総合情報システム | 国土交通省のデータを使って，不動産の価格予測を行うシステムを作ることにします。
条件
- 東京都の中古マンションのみを対象とする
- リノベーションされた物件は対象外とする（築年数の持つ意味が変わるため）
APIに送信する特徴量
以下の特徴量を送ることにします。
| 特徴量 | 説明 | データ型 |
|---|---|---|
| address | 市区町村レベルまでの住所 | string |
| area | 専有面積 | int or float |
| building_year | 竣工年 | int or float |
入力（request）/出力（response）
例えば予測のリクエストを
{ "address": "東京都千代田区", "area": 30, "building_year": 2013 }
のようにして送ると，
{ "status": "OK", "preidcted": 40000000 }
のような値を返すAPIとします。
予測モデルを作る
データの取得
土地総合情報システム | 国土交通省のAPIを使います。
import requests import json import pandas as pd import os url = "https://www.land.mlit.go.jp/webland/api/TradeListSearch" # 東京都，2005Q3 ~ 2019Q3のデータ（DLに10分ほどかかるので注意） payload = {"area": 13, "from": 20053, "to": 20193} response = requests.get(url, params=payload) data = json.loads(response.text) df = pd.DataFrame(data["data"]) # 保存 os.mkdir("input") df.to_csv("input/raw.csv", index=False)
基礎的な前処理
まず基礎的な前処理を行い，APIで受け取るデータと同様の状況にしていきます。
import pandas as pd df = pd.read_csv("input/raw.csv") # 使用するデータの選択 ---------------------------- # マンションのみを対象にする is_mansion = df["Type"] == "中古マンション等" df = df.loc[is_mansion, :] # リノベーションされた物件は対象外とする is_not_renovated = df["Renovation"] != "改装済" df = df.loc[is_not_renovated, :] # 列名変更 ---------------------------------------- df = df.rename(columns={"TradePrice": "price", "Area": "area"}) # 特徴量の生成 ------------------------------------ # 住所 df["address"] = df["Prefecture"] + df["Municipality"] # 竣工年の和暦を西暦にする years = df["BuildingYear"].str.extract(r"(?P<period>昭和|平成|令和)(?P<year>\d+)") years["year"] = years["year"].astype(float) years["period"] = years["period"].map({"昭和": 1925, "平成": 1988, "令和": 2019}) df["building_year"] = years["period"] + years["year"] # apiが利用される場面を考えて四半期を月に変更 years = df["Period"].str.extract(r"(\d+)")[0] zen2han = {"１": "1", "２": "2", "３": "3", "４": "4"} quarters = df["Period"].str.extract(r"(\d四半期)")[0]\ .str.replace("四半期", "").map(zen2han).astype(int) months = (quarters * 3 - 2).astype(str) df["trade_date"] = pd.to_datetime(years + "-" + months) # 使用する変数の取り出し cols = ["price", "address", "area", "building_year", "trade_date"] df = df[cols].dropna() # 保存 -------------------------------------------- df.to_csv("input/basic_data.csv", index=False)
こんな感じになります。
- priceは目的変数
- address, area, building_yearは予測のときはAPIの利用者が入力する
- trade_dateは予測のときは「APIが利用された日」を使う（システムが作る特徴量）
という想定です。
前処理関数の定義
addressは文字列，trade_dateはdatetimeか文字列なので，このまま機械学習モデルに入れるわけにはいきません。
今回はLightGBMを使うので完全にカテゴリカル変数であるaddressはLightGBM内でcategoricalにすればいいとしても，trade_dateは順序があるカテゴリカル変数なので数値にしたいところです。
そんなちょっとした特徴量の加工をするクラスを定義します1
from sklearn.base import BaseEstimator, TransformerMixin import pandas as pd import numpy as np class skPlumberBase(BaseEstimator, TransformerMixin): """Pipelineに入れられるTransformerのベース""" def __init__(self): pass def fit(self, X, y=None): return self def transform(self, X): return self class Date2Int(skPlumberBase): def __init__(self, target_col): self.target_col = target_col def transform(self, X): """unix時間に変換する""" dates = pd.to_datetime(X[self.target_col]).astype(np.int64) / 10**9 X[self.target_col] = dates.astype(int) return X class ToCategorical(skPlumberBase): """LightGBMにcategoryだと認識させるため， カテゴリカル変数をpandas category型にする """ def __init__(self, target_col): self.target_col = target_col def transform(self, X): X[self.target_col] = X[self.target_col].astype("category") return X
Date2Intは
0 2019-07-01
1 2018-10-01
2 2018-07-01
3 2018-04-01
4 2018-04-01
...
137548 2008-01-01
137549 2007-10-01
137550 2007-10-01
137551 2007-07-01
137552 2007-04-01
Name: trade_date, Length: 137553, dtype: object
を
0 1561939200
1 1538352000
2 1530403200
3 1522540800
4 1522540800
...
137548 1199145600
137549 1191196800
137550 1191196800
137551 1183248000
137552 1175385600
Name: trade_date, Length: 137553, dtype: int32
のようにするものです。
unix時間にすると桁数が増えて使用メモリが増えて効率的ではない気がしますが，あくまで例ということで…
学習
前処理して，学習して，前処理パイプラインと予測モデルをpickleで保存します。
from sklearn.pipeline import Pipeline from pipeline import Date2Int, ToCategorical import pandas as pd import pickle import lightgbm as lgb from sklearn.model_selection import train_test_split # データ読み込み df = pd.read_csv("input/basic_data.csv") y = df["price"] X = df.drop("price", axis=1) # 前処理パイプラインの定義 preprocess = Pipeline(steps=[ ("date_to_int", Date2Int(target_col="trade_date")), ("to_category", ToCategorical(target_col="address")) ], verbose=True) # 前処理 X = preprocess.transform(X) # データを分割 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42) # 学習 params = { "n_estimators": 100_000, "min_child_samples": 15, "max_depth": 4, "colsample_bytree": 0.7, "random_state": 42 } model = lgb.LGBMRegressor(**params) model.fit(X_train, y_train, eval_metric="rmse", eval_set=[(X_test, y_test)], early_stopping_rounds=100) print("best scores:", dict(model.best_score_["valid_0"])) # 保存 pickle.dump(preprocess, open("preprocess.pkl", "wb")) pickle.dump(model, open("model.pkl", "wb"))
testデータに対するRMSEが1950万円くらいあります。これほど少ない特徴量だとさすがにひどい精度になりますね。
best scores: {'rmse': 19500026.074094355, 'l2': 380251016890359.75}
APIを作る
テストを書く
作るものを決める
これから作るAPIは，予測したい物件の特徴量を
{ "address": "東京都千代田区", "area": 30, "building_year": 2013 }
のようにJSONにして送ると，
{ "status": "OK", "preidcted": 40000000 }
テストを書く
import unittest import requests import json class APITest(unittest.TestCase): URL = "http://localhost:5000/api/predict" DATA = { "address": "東京都千代田区", "area": 30, "building_year": 2013 } def test_normal_input(self): # リクエストを投げる response = requests.post(self.URL, json=self.DATA) # 結果 print(response.text) # 本来は不要だが，確認用 result = json.loads(response.text) # JSONをdictに変換 # ステータスコードが200かどうか self.assertEqual(response.status_code, 200) # statusはOKかどうか self.assertEqual(result["status"], "OK") # 非負の予測値があるかどうか self.assertTrue(0 <= result["predicted"]) if __name__ == "__main__": unittest.main()
※あくまで例です。実際にちゃんと作るときはもっと沢山（正常系だけでなく異常系も）テストケースを作ります。
Flaskでアプリを作成
FlaskはPythonの軽量なWebフレームワーク（Webアプリを簡単に作ることができるライブラリ）で，極めて少ないコード量でアプリを作ることができます。
「機械学習モデルを動かすだけ」みたいな単純な動作をするAPIには最適なフレームワークです。
以下のように書いていきます。
from flask import Flask, request, jsonify, abort import pandas as pd import pickle from datetime import datetime import sys sys.path.append("./model") # 前処理で使った自作モジュール「pipeline」を読み込むためPYTHONPATHに追加 app = Flask(__name__) # アプリ起動時に前処理パイプラインと予測モデルを読み込んでおく preprocess = pickle.load(open("model/preprocess.pkl", "rb")) model = pickle.load(open("model/model.pkl", "rb")) @app.route('/api/predict', methods=["POST"]) def predict(): """/api/predict にPOSTリクエストされたら予測値を返す関数""" try: # APIにJSON形式で送信された特徴量 X = pd.DataFrame(request.json, index=[0]) # 特徴量を追加 X["trade_date"] = datetime.now() # 前処理 X = preprocess.transform(X) # 予測 y_pred = model.predict(X, num_iteration=model.best_iteration_) response = {"status": "OK", "predicted": y_pred[0]} return jsonify(response), 200 except Exception as e: print(e) # デバッグ用 abort(400) @app.errorhandler(400) def error_handler(error): """abort(400) した時のレスポンス""" response = {"status": "Error", "message": "Invalid Parameters"} return jsonify(response), error.code if __name__ == "__main__": app.run(debug=True) # 開発用サーバーの起動
前節で書いたテストを走らせるとこうなります
$ python3 api_test.py { "predicted": 45833222.1903707, "status": "OK" } . ---------------------------------------------------------------------- Ran 1 test in 0.015s OK
期待通り，predictedとstatusが返っているようです。
