

# 目的

株取引をしている友人から移動平均線を使った取引手法が実際に通用しているのか検証したいという相談を受けました。彼はあまりプログラミング関連の知識はありませんが、問題をきちんと数式で記述できます。コードの勉強をしてからようやく彼の問題意識の検証、ではあまりに時間がもったいないです。

数式に近い形でクエリの条件をかけるような検索ツールがあれば、SQLやExcelを扱えなくても簡単なPythonが書ければすぐに分析に取り掛かれます。また、コードの可読性も大幅に向上します。

というわけで最低限の機能ですがデータ検索のためにQuerypyを公開しました。
[https://github.com/yaufai/querypy](https://github.com/yaufai/querypy)

# 簡単な使い方

おそらく例をみれば使い方はすぐ分かるかと思いますので、READMEの簡単な日本語訳を載っけます。
例えば下のようなデータから成績が条件を満たす（合格点に達しているか、など）学生を知りたいとしましょう。
```csv
名前, 解析入門, 経済学入門
山田,      80,      50
　劉,      50,      80
```

検索用のオブジェクトとクエリを作成すればすぐに結果を得ることができます。
```py
from querypy import Querypy

data    = Querypy.read_csv("scores.csv", "名前")
math101 = data.get_term("解析入門")
econ101 = data.get_term("経済学入門")

pass_math_exam  = math101 >=  60
better_in_econ  = math101 < econ101

print(data.find(pass_math_exam))
# -> ['劉']
print(data.find(better_in_econ))
# -> ['山田']
```

論理積や論理和を使った複雑な式にも対応できます。
```py
print(data.find(
    (econ101 == 100) & (math101 == 100)
))
# -> []　（全教科満点はいなかった）
print(data.find(
    (econ101 >= 60) | pass_math_exam
))
# -> ['山田', '劉']  （どれか1つに合格した人は2人でした）
```

# 実装にあたって

本来`==`はオブジェクトが同じものかを比較しますが、`math101 == 100`は`bool`値としては評価されていません。これを可能にしているのはPythonの演算子の振る舞いを上書きしているからです。

例えば`==`では
```py
    def __eq__(self, term: Term) -> Formula.Formula:
        return self.__abstract_operator__(term, Formula.Relations.Equal)
```
というメソッドで上書きしています。型アノテーションをみると分かるように、返り値は`bool`ではなく`Formula.Formula`となっています。このオブジェクトがCSVの項が条件を満たしているかの判断に責任を持っているわけです。

# 今後、やっていきたいこと

現在このツールはまだまだ最低限の機能（いや、それよりも少ない）しか持っていません。自分で使うにもまだまだ不自由なので追加していきたい機能として以下のものを考えています。

読者の方でもしこのソフトウェアに貢献されたい方が見えましたらぜひご協力ください。下にないものでも大歓迎です！

## ユニットテストを書く

手元である程度動くことを確認した程度で本来公開するのに必要な水準さえ確認できていません。

## 数値以外にも対応

現在は主キーを除いて全ての要素は数値しか取ることができません。例えば「経済学を専攻していてEcon101で80点以上」のようなクエリは（わざわざ専攻に数値をわり当てるなどの工夫をしなければ）対応できません。

## JSONなどに対応

基本的にはCSVファイルだけでも多くのことが表現できていますが、他にデータを記述する形式があるので、そちらの方も視野に入れていきたいです。

## 関数をつかった表現

現在は要素の比較までしか対応できてません。でもデータ分析には本来もっと複雑な条件が関わってくるはずです。

例えば上の例で「二教科の合計点が100点以上」というようなクエリで絞り込みたいときに合計を求める自作関数`compute_total_score`をみたいにかけるといいなと思っています：
```py
def compute_total_score(terms: List[Term]) -> Union[float, int]:
    return reduce(lambda acc, cur: acc + cur, terms, 0)

query = compute_total_score([math101, econ101]) >= 100
```

## 二項間の関係

現在は、各行ごとに条件をチェックしています。学生の成績のように個別に評価する場合はそれで十分ですが、時系列データで「昨日は条件Aをみたし、今日は条件Bをみたす日付の組すべて」みたいな検索はできません。

## 2つのCSVを利用

関係モデルのデータベースのような構造をもった複数のCSVが扱えればようやく多くの実務で使える検索ツールとしてはじめの一歩を踏み出せたといえるでしょう。


# バージョン0.0.1

先日公開した[Querypy](https://github.com/yaufai/querypy)で
1. テスト環境の整備
2. 関数記号の導入

までやったのでバージョンを0.0.1にしました。

## テスト環境

[momotaro98](https://github.com/momotaro98)さんがutilsのテストを書いてくださったついでに、READMEの開発指南にテスト関係の項目を追加しました。今後は重要な機能であったり複雑なロジックを実装した部分を意識しつつテスト済みコンポーネントを増やしていきたいです。

## 細かいREADMEの修正

READMEのスクリプトが間違っていたので修正しました。

## 項の間の演算

加減乗除とべき乗をつかって計算できるようになりました。

例によって以下の`scores.csv`を読み込んで条件に合う学生を探し出しましょう。
```csv
名前, 解析入門, 経済学入門
山田,      80,      50
　劉,      50,      80
```
```py
data    = Querypy.read_csv("scores.csv", "名前")
math101 = data.get_term("解析入門")
econ101 = data.get_term("経済学入門")
```

例えば二教科`math101`, `econ101`の合計点数は`math101 + econ101`と表現して実際に条件式に組み入れることができます。

```py
data.find(math101 + econ101 >= 100)
# -> ['山田', '劉']
```

他にも加減乗除、べき乗すべてが同様に条件式のなかで扱えることになります。

```py
# 解析入門のほうが点数が高い人
math101 - econ101 > 0

# 数学の点数を1.5倍の重みで評価する場合
math101 * 1.5 + econ101 >= 120

# 経済学入門のほうが１０％以上点数が高い人
econ101/math101 >= 1.1

# 両方共50点との距離を測る
(math101 - 50) ** 2 + (econ101 - 50) ** >= 100
```

## 関数記号の導入

関数記号を導入することでレコードに任意の演算を施した結果をつかって条件を記述できるようになりました。

例えばGPAを以下のようなルールでつけて、GPA３以上の学生を見つけたいとしましょう：

* 60点以上69点以下：２点
* 70点以上79点以下：３点
* 80点以上　　　　：４点

この条件分岐を含んだ式は上のような加減乗除べき乗だけでは表現できませんが、

```py
def get_grade_point(test_result):
    if test_result >= 80:
        return 4
    elif test_result >= 70:
        return 3
    elif test_result >= 60:
        return 2
    else:
        return 0

def get_gpa(math, econ):
    return (get_grade_point(math) + get_grade_point(econ)) / 2
```

という関数を用意すれば`get_gpa(math101, econ101)`みたいな感じで計算できそうです。これを条件式に組み込みたいときはまず

```py
from querypy.FunctionSymbol import FunctionSymbol

gpa_func_sym = FunctionSymbol(get_gpa)
```

という形で関数記号オブジェクト（`FunctionSymbol`）を作成します。
そしてそのあとは通常どおりに条件式を不等号をつかって作成できます。

```py
data.find(gpa_func_sym(math101, econ101) >= 3)
```

# 今後の方針

## ２つのレコードを利用

順序付きデータや時系列データでは
「前の項が条件Aをみたして次の項が条件Bをみたすもの」
「一時間での差分が５％以内」
など複数レコードにまたがる条件が使えると便利です。

まだこの機能の構想、設計ができていないのですぐにコードに落とし込むことはできませんが、逆になにかアイデアがあればIssueいただけると嬉しいです。

## テストの充実

せっかくテストのルールを作ることができたのでそれを活かして製品の質を上げていきたいです。

## 対象レコードの範囲

今はかなりかぎられた条件を満たすCSVしか使うことができません。
より一般的なCSVでも検索できるよう、またJSONなどほかの形式でも扱えるようにデザインを変更していきたいです。



# マルチスレッドのメリット：並列化で速度改善

複数の処理を同時に実行することによって繰り返し処理全体にかかる時間を短縮することができます。

ただ、マルチスレッドを使えば計算資源を多く利用できるようになるというわけではありません。なので、計算量のある処理を並行して行うというよりは、IOやHTTP APIコールのオーバヘッドで計算していない時間が発生するような処理のほうが向いています。例えば１秒まつという処理をする関数を１０回実行するとき、逐次実行していけばおよそ十秒かかりますが、１０スレッドを準備してそれぞれが独立に１秒待つことで全体としてはおよそ１秒で終えることができます。


ただし特別な制御をしない限りは、並列に行われる仕事の順序を決めることはできませんので、その辺りの要求がある場合には単純な逐次実行が優れているでしょう。


マルチスレッドによって時間が早くなったかをみるための時間の測定には`time`モジュールを使います。


```py
import time

time_start = time.time()

処理

print(time.time() - time_start)
```

マルチスレッドそのものは`threading`モジュールにある`Thread`クラスを活用して作成します。


# もっとも単純な並列処理

もっとも単純な「１秒待つ」処理を複数スレッドを利用して並列に実行し、実行時間が実際に短縮されていることを確認します。

```py
def wait_one_sec(i):
    time.sleep(1)
```

並列化の全体像は以下の通りです。最後に個々のスレッドの終了を待ちます（`Thread.join`）。

```py
threads = []

for i in データ:
    thread = threading.Thread(target=データを利用した仕事, args=(i,))
    thread.start()
    threads.append(thread)

for thread in threads:
    thread.join()
```

[こちらで先ほどの一秒まつ関数を仕事としてマルチスレッドありなしを試した](https://github.com/yaufai/notebooks/blob/a6b72dcead96f101b8d48eb419c54bc7ba8c98cd/python/multithreading/basic-time-measure.py)ところ以下のようになりました。予想通り本来なら１０秒強かかる仕事を全体としては１秒強で完了できます。


```
$ python3 basic-time-measure.py 
マルチスレッドなし:  10.012403964996338
マルチスレッドあり:  1.005929708480835
```

# 一つのキューからデータを取り出していく

スレッド間である程度情報を共有してスレッド全体をある程度制御したい場合には（例えば全体のスレッド数を制限したい場合）、キューというデータ構造を利用することで実現できます。


1. キューを準備する
1. キューの先頭に対して処理する関数をつくる
1. キューと仕事する関数を使ってスレッドを作成する
1. スレッドを起動してキューにデータを追加する


仕事する関数は、データではなくキューを受け取り、関数のなかでキューからデータを取り出すようにします。

```py
def wait_one_sec(queue_obj):
    item = queue_obj.get()
    time.sleep(1)
    queue_obj.task_done()
```

## 参考：キューのAPI

キュー構造は`queue`モジュールにある`queue.Queue`によって作成できます。


* `Queue.put`: キューの末尾にデータを追加
* `Queue.get`: キューの先頭データを取得
    * デフォルトではキューにデータがない場合、追加されるまでブロックする
* `Queue.join`: キューにあるデータがすべて取り除かれて処理が完了するまでブロックする
    * `Queue.task_done`: `Queue.get`で取り出したデータについて処理が完了したことを示す


## キューを利用するマルチスレッドの全体像


スレッドの関数がキューを受け取るようになるので、受け渡すデータもキューになります。またすべての仕事を完了させたいときには`Queue.join`ですべてのキュー上の要素に`task_done`が呼ばれるのを待ちます。

```py
対象データのキュー = queue.Queue()

for i in data:
    thread = threading.Thread(target=仕事をする関数, args=(対象データのキュー,))
    thread.start()
    q.put(i)

q.join()
```


[ここに先ほどの１秒待つ関数をあてがったもの](https://github.com/yaufai/notebooks/blob/a6b72dcead96f101b8d48eb419c54bc7ba8c98cd/python/multithreading/use-queue.py)で実際に速度を計測すると、１秒ほどで全処理が完了し、十分に並列処理ができていることが分かります。

```
$ python3 use-queue.py 
1.0044941902160645
```

# 同時に処理するスレッド数を制限する

スレッド数が制限範囲を超えているときに、キューにある処理が完了するまでブロック（`join`）するとスレッド数を制限して並列処理をすることができます。もちろん無制限にスレッドを作成できたときに比べると全体としては遅くなりますが、２スレッドつかえば１スレッドだけでやった時の半分の時間で、$n$スレッドつかえば、全体としては$1/n$の時間で仕事を終えることができます。



全体像はこのようになります。すでに本体のスレッドが生成されているので、`threading.active_count`の返り値は自分で明示的に作成したスレッド数よりも１大きくなることに注意します。


```py

対象データのキュー = queue.Queue(最大スレッド数)

for i in data:
    thread = threading.Thread(target=仕事をする関数, args=(対象データのキュー,))
    thread.start()
    q.put(i)

    if threading.active_count() >= 最大スレッド数 + 1:
        q.join()
```




同じように[１秒まつメソッド１０回を最大スレッド数ごとに実行](https://github.com/yaufai/notebooks/blob/a6b72dcead96f101b8d48eb419c54bc7ba8c98cd/python/multithreading/use-queue-maximum.py)した結果です。

```
$ python3 use-queue-maximum.py 
サイズ1:  10.020220518112183
サイズ2:  5.010011434555054
サイズ5:  2.0056521892547607
```

かかる時間がつかえたスレッド数で割った値になることが確認できます。


# 参考にさせていただいた資料


* [https://news.mynavi.jp/article/python-32/](https://news.mynavi.jp/article/python-32/)


分かりやすくまとまっていてマルチスレッドの基本的な仕組みと実装を把握するのに最適です。


* [https://teratail.com/questions/49684](https://teratail.com/questions/49684)

[yohhoy](https://teratail.com/users/yohhoy)さんの回答がキューを使ったマルチスレッド処理の実装を理解するのに役立ちました。