乙Py先生のプログラミング教室
初学者のためのプログラミング学習サイト

仮想通貨

今回はブロックチェーンのプログラムを
みていきたいと思います。

解説動画はこちら


 
今回は「ブロックチェーン」についてです。


ブロックチェーンとは

分散型台帳技術の一種であり、データを安全かつ
改ざん不可能な形で記録するための仕組みのことです。

連続するブロックで鎖のように構成されています
(参考:NTTデータ)

fig_02_01


最初のブロックは、ジェネシスブロックと呼ばれています。

各ブロックには次のような情報が格納されています。
・トランザクションデータ:取引の詳細情報
・タイムスタンプ:取引が行われた日時
・前のブロックのハッシュ:直前のブロックの暗号学的ハッシュ値
・現在のブロックのハッシュ:現在のブロックのデータから生成されたハッシュ値


ハッシュとは

取引情報やタイムスタンプなどから計算される数値のことです。
ハッシュ計算にはsha256(Secure Hash Algorithm 256-bit)などが使われています。

このハッシュ値を前のブロックの物と比較して
整合性をチェックしているので、改竄が難しくなっています。

ブロック1:データ + ハッシュ0(ジェネシスブロックのハッシュ)
ブロック2:データ + ハッシュ1
ブロック3:データ + ハッシュ2

チェーンの途中のブロックのデータを改竄すると
それ以降全てのハッシュ値を改竄しないといけないため
ものすごい難易度になります。


ブロックチェーンの簡単なプログラム

ブロックチェーンを作ってブロックを繋げるだけの
簡単なサンプルです。
import hashlib
import time

# ブロックを定義するクラス
class Block:
    def __init__(self, index, previous_hash, timestamp, data):
        self.index = index
        self.previous_hash = previous_hash
        self.timestamp = timestamp
        self.data = data
        self.hash = self.calculate_hash()

    def calculate_hash(self):
        sha = hashlib.sha256()
        sha.update((str(self.index) + str(self.previous_hash) + str(self.timestamp) + str(self.data)).encode('utf-8'))
        return sha.hexdigest()

# ブロックチェーンを定義するクラス
class Blockchain:
    def __init__(self):
        self.chain = [self.create_genesis_block()]

    def create_genesis_block(self):
        return Block(0, "0", int(time.time()), "Genesis Block")

    def get_latest_block(self):
        return self.chain[-1]

    def add_block(self, new_block):
        new_block.previous_hash = self.get_latest_block().hash
        new_block.hash = new_block.calculate_hash()
        self.chain.append(new_block)

# 最初のブロックチェーンの初期化
my_blockchain = Blockchain()

# 最初のブロックチェーンに新しいブロックを追加
my_blockchain.add_block(Block(1, "", int(time.time()), "Block 1 Data"))
my_blockchain.add_block(Block(2, "", int(time.time()), "Block 2 Data"))

# ブロックチェーンの内容を表示
for block in my_blockchain.chain:
    print(f"Index: {block.index}")
    print(f"Previous Hash: {block.previous_hash}")
    print(f"Timestamp: {block.timestamp}")
    print(f"Data: {block.data}")
    print(f"Hash: {block.hash}")
    print("-" * 30)
Index: 0
Previous Hash: 0
Timestamp: 1724488836
Data: Genesis Block
Hash: 042bf4db73e03d3047fe19aee7a9d5e4d95a446f8420f56704d2084f231f1350
------------------------------
Index: 1
Previous Hash: 042bf4db73e03d3047fe19aee7a9d5e4d95a446f8420f56704d2084f231f1350
Timestamp: 1724488836
Data: Block 1 Data
Hash: 3cff362cc92bcf78e7d78f0f01435ab18e17d0f543b57634953ee163b6d344f3
------------------------------
Index: 2
Previous Hash: 3cff362cc92bcf78e7d78f0f01435ab18e17d0f543b57634953ee163b6d344f3
Timestamp: 1724488836
Data: Block 2 Data
Hash: bbd7cdf536222d1b757061d459dd7c240dfef1eb311de3b1f5e1ca9f6dfae37e


ブロックの作成時にはHash値が計算され
ブロックには前のブロックのHash値も格納されています。


次は実際の取引っぽいデータでブロックを作り
ブロックチェーンの有効性も検証してみます。
import hashlib
import time
from typing import List

class Block:
    def __init__(self, index: int, previous_hash: str, timestamp: int, transactions: List[dict]):
        self.index = index
        self.previous_hash = previous_hash
        self.timestamp = timestamp
        self.transactions = transactions
        self.hash = self.calculate_hash()

    def calculate_hash(self):
        block_string = f"{self.index}{self.previous_hash}{self.timestamp}{self.transactions}"
        return hashlib.sha256(block_string.encode()).hexdigest()

class Blockchain:
    def __init__(self):
        self.chain = [self.create_genesis_block()]

    def create_genesis_block(self):
        return Block(0, "0", int(time.time()), [{"sender": "genesis", "receiver": "network", "amount": 0}])

    def get_latest_block(self):
        return self.chain[-1]

    def add_block(self, transactions: List[dict]):
        latest_block = self.get_latest_block()
        new_block = Block(latest_block.index + 1, latest_block.hash, int(time.time()), transactions)
        self.chain.append(new_block)

    def is_chain_valid(self):
        for i in range(1, len(self.chain)):
            current_block = self.chain[i]
            previous_block = self.chain[i-1]
            print(f"current_block.hash : {current_block.hash} , current_block.calculate_hash : {current_block.calculate_hash()}")
            print(f"current_block.previous_hash : {current_block.previous_hash} , previous_block.hash : {previous_block.hash}")
            if current_block.hash != current_block.calculate_hash():
                return False
            if current_block.previous_hash != previous_block.hash:
                return False
        return True

# ブロックチェーンの初期化
my_blockchain = Blockchain()

# 取引の例
transactions1 = [
    {"sender": "Alice", "receiver": "Bob", "amount": 50},
    {"sender": "Bob", "receiver": "Charlie", "amount": 30}
]
my_blockchain.add_block(transactions1)

transactions2 = [
    {"sender": "Alice", "receiver": "Charlie", "amount": 20},
    {"sender": "Charlie", "receiver": "Bob", "amount": 10}
]
my_blockchain.add_block(transactions2)

# ブロックチェーンの内容を表示
for block in my_blockchain.chain:
    print(f"Index: {block.index}")
    print(f"Previous Hash: {block.previous_hash}")
    print(f"Timestamp: {block.timestamp}")
    print(f"Transactions: {block.transactions}")
    print(f"Hash: {block.hash}")
    print("-" * 30)

# ブロックチェーンの有効性を検証
print("Is blockchain valid?", my_blockchain.is_chain_valid())
Index: 0 Previous Hash: 0 Timestamp: 1724483719 Transactions: [{'sender': 'genesis', 'receiver': 'network', 'amount': 0}] Hash: ec68f33868e558eca7f3fd6260101430eb12e7e8eb9b328a18882ef6f3ed229c ------------------------------ Index: 1 Previous Hash: ec68f33868e558eca7f3fd6260101430eb12e7e8eb9b328a18882ef6f3ed229c Timestamp: 1724483719 Transactions: [{'sender': 'Alice', 'receiver': 'Bob', 'amount': 50}, {'sender': 'Bob', 'receiver': 'Charlie', 'amount': 30}] Hash: 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 ------------------------------ Index: 2 Previous Hash: 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 Timestamp: 1724483719 Transactions: [{'sender': 'Alice', 'receiver': 'Charlie', 'amount': 20}, {'sender': 'Charlie', 'receiver': 'Bob', 'amount': 10}] Hash: b4e7a335791d6b846fba35dece29095f8c2792cde02b6a012403ab6322cd1132 ------------------------------ current_block.hash : 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 , current_block.calculate_hash : 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 current_block.previous_hash : ec68f33868e558eca7f3fd6260101430eb12e7e8eb9b328a18882ef6f3ed229c , previous_block.hash : ec68f33868e558eca7f3fd6260101430eb12e7e8eb9b328a18882ef6f3ed229c current_block.hash : b4e7a335791d6b846fba35dece29095f8c2792cde02b6a012403ab6322cd1132 , current_block.calculate_hash : b4e7a335791d6b846fba35dece29095f8c2792cde02b6a012403ab6322cd1132 current_block.previous_hash : 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 , previous_block.hash : 3a39a53e1423ece90ce7418689a3593959f3041e5b5c5ae7ddf3f95ec9264d25 Is blockchain valid? True

現在のブロックのハッシュと計算されたハッシュ値
現在のブロックにある前のハッシュ値と前のブロックのハッシュ値
で整合性を見ています。





暗号通貨のマイニングの仕組み

マイニング(採掘)とは、分散型ネットワークにて新しい取引を検証し
それをブロックチェーンに追加するプロセスのことです。

特にビットコインなどのProof of Work(PoW)ベースの
暗号通貨において重要な役割を果たしています。

コンセンサスアルゴリズムと呼ばれる
新しいブロックの追加に合意するための仕組みが備わっており
ビットコインなどではProof of Work (PoW)と呼ばれるものが使われています。

Proof of Work (PoW):
計算能力を競い合い、最初に問題を解いたノードが
ブロックを追加する権利を得るというものです。
計算には莫大なリソースが必要になります。


マイニングの基本的な仕組み

ここではすごい荒く説明です。

1.取引の収集と検証:
各取引が有効であることを確認するために
デジタル署名や残高の確認などの検証が行われる

2.ブロックの形成:
検証された取引は、一定数集まると1つのブロックにまとめられる
ブロックには、取引情報の他に、前のブロックのハッシュ値、タイムスタンプ
ナンス(nonce)という特別な値が含まれる。

3.ハッシュ計算とナンスの探索:
マイナーは、ブロックヘッダー(ブロックの要約情報)に
対してハッシュ関数を適用し、ハッシュ値を計算する

目的は、そのハッシュ値が特定の条件(例えば、ハッシュ値が一定の数の先頭ゼロを持つこと)
を満たすようにすること

ナンスはこの条件を満たすために調整される値で
マイナーはナンスを変化させながら何度もハッシュ計算を繰り返します。

4.ブロックの追加:
条件を満たすハッシュ値が見つかると、そのブロックが承認され
ブロックチェーンに追加される

このプロセスを「プルーフ・オブ・ワーク」と呼び
計算量の多さがブロックの正当性を保証する

5.報酬の獲得:
ブロックを成功裏に追加したマイナーには
暗号通貨の新規発行分(ブロック報酬)や取引手数料などの報酬が与えられる

ビットコインの場合、最初のブロック(ジェネシスブロック)以降
約4年ごとにブロック報酬は半減する仕組み(半減期)が導入されている



マイニングのコード

ハッシュ計算とナンスの探索部分を難易度を下げて
実験してみます。


ハッシュを計算する際に
先頭に0が難易度の数値分並ぶかどうかを目的にしています。
0が並ぶまで計算を続け、ヒットしたらハッシュを追加します。

難易度の数値を上げると時間が掛かるようになります。


import hashlib
import time

class Block:
    def __init__(self, index, previous_hash, timestamp, data, nonce=0):
        self.index = index
        self.previous_hash = previous_hash
        self.timestamp = timestamp
        self.data = data
        self.nonce = nonce
        self.hash = self.calculate_hash()

    def calculate_hash(self):
        value = (str(self.index) + str(self.previous_hash) + str(self.timestamp) +
                 str(self.data) + str(self.nonce)).encode()
        return hashlib.sha256(value).hexdigest()

    def mine_block(self, difficulty):
        target = '0' * difficulty
        while self.hash[:difficulty] != target:
            self.nonce += 1
            self.hash = self.calculate_hash()
        print(f"Block mined: {self.hash}")

class Blockchain:
    def __init__(self):
        self.chain = [self.create_genesis_block()]
        self.difficulty = 5  # 難易度

    def create_genesis_block(self):
        return Block(0, "0", time.time(), "Genesis Block")

    def get_latest_block(self):
        return self.chain[-1]

    def add_block(self, new_block):
        new_block.previous_hash = self.get_latest_block().hash
        new_block.mine_block(self.difficulty)
        self.chain.append(new_block)

# ブロックチェーンの初期化
blockchain = Blockchain()

# 新しいブロックを追加
print("Mining block 1...")
new_block = Block(1, "", time.time(), "Block 1 Data")
blockchain.add_block(new_block)

print("Mining block 2...")
new_block = Block(2, "", time.time(), "Block 2 Data")
blockchain.add_block(new_block)

print("Mining block 3...")
new_block = Block(3, "", time.time(), "Block 3 Data")
blockchain.add_block(new_block)
Mining block 1...
Block mined: 00000b1dcfcdb93abbe9d2656e7c36851aae271bfcd45aecb3ec1a0f43474228
Mining block 2...
Block mined: 0000083671360216d051d4ca6a2fef3fc783e311ec3540a484a24a86c85471cb
Mining block 3...
Block mined: 000001d4692c5f50c4e90b74ce7a0bf87839de5893667b424846b8df1c7e5db4


なんとなくではありますが
マイニングの大変さが分かるかと思います。



まとめ

ブロックチェーン技術は暗号通貨の取引や
透明性、セキュリティ、分散化の特性により
今後さらに多くの分野での活用が期待されています。

正当性の証明や、改竄防止関連の仕事が
増えてくると予想しているので
覚えておくと使う機会が来るかもしれません。

それでは





今回は仮想通貨の自動売買ロジックの
組み立て方のチュートリアルコードを
動かしてみました。


解説動画はこちら



はじめに


今回は書籍
「日給300万円のSS級トレーダーが明かすbotterのリアル」
のチュートリアルコードを動かす方法などについてです。


書籍画像
No description has been provided for this image


チュートリアルのコードはこちら
githubのコード

こちらのコードは
そのままでは動かない可能性もあるので
それ含めて動かせる様にしてみました。


チュートリアル環境の構築


1.Dockerを用いる方
Readme通りにやるだけです。
# コードのクローン
git clone https://github.com/richmanbtc/mlbot_tutorial.git

# Dockerの起動
cd mlbot_tutorial
docker-compose up -d

# Jupyterの起動
# http://localhost:8888 をブラウザで開く

2.自分のJupyter環境 or Colabでやる方
必須ライブラリをインストールしておく必要があります。
# Jupyterの人は全部
ccxt : 仮想通貨取引用のライブラリ
TA-lib : 指標の計算ライブラリ

# Colabの人は不要
numba : Pythonの高速化ライブラリ
scikit-learn : 機械学習用ライブラリ
lightGBM : 機械学習のモデル構築ライブラリ

なおインストール方法は割愛します。
インストールが自力で進める方だけ
進んでください。


チュートリアルの進め方


ここからはチュートリアルのコードを
そのまま進める形になります。

チュートリアルのコードを
そのまま実行してみて下さい。

チュートリアルはこのような
内容になっていました。

1.ライブラリのインポート
2.データ取得
3.データ読み込み
4.前処理
5.目的変数の計算
6.モデルの学習
7.バックテスト


このうち
2.データ取得
3.データ読み込み
に関しては環境によっては
うまく行えない可能性がありました。

ライブラリの差異などにより
うまくデータが生成されませんでした。

なので代わりのコードを
置いておきます。

データ取得のコード
import urllib.request
import os
import time
from datetime import datetime, timedelta
# SSLの問題が有ったら追加
import ssl
ssl._create_default_https_context = ssl._create_unverified_context

def get_date_range(start_date, end_date):
    start = datetime.strptime(start_date, '%Y-%m-%d')
    end = datetime.strptime(end_date, '%Y-%m-%d')
    date_range, current_date= [],start
    while current_date <= end:
        date_range.append((current_date.year, current_date.month, current_date.day))
        current_date += timedelta(days=1)
    return date_range

def get_data(market,start_date,end_date,data_dir = "data"):
    data_dir = f"{data_dir}/{market}"
    url_base = 'https://api.coin.z.com/data/trades/{0}/{1}/{2:02}/{1}{2:02}{3:02}_{0}.csv.gz'
    dates = get_date_range(start_date, end_date)
    for d in dates:
        year,month,day = d
        url = url_base.format(
            market,
            year,
            month,
            day
        )
        file_name = os.path.basename(url)
        if not os.path.exists(f"{data_dir}/{year}"):
            os.makedirs(f"{data_dir}/{year}")
        save_path = os.path.join(f"{data_dir}/{year}", file_name)
        urllib.request.urlretrieve(url, save_path)
        time.sleep(1.37)
# マーケットと開始日、終了日を指定
market = "BTC_JPY"
start_date = "2024-01-01"
end_date = "2024-02-22"
get_data(market,start_date,end_date)

実行するとこのコードが置いてある
ディレクトリ配下にdataが配置されます。

データの読み込み
import os
import glob

def make_df(file_path,interval_sec):
    df = pd.read_csv(file_path)
    df = df.rename(columns={'symbol': 'market',})
    df['price'] = df['price'].astype('float64')
    df['size'] = df['size'].astype('float64')
    df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)
    df['side'] = np.where(df['side'] == 'BUY', 1, -1).astype('int8')
    df['timestamp'] = df['timestamp'].dt.floor('{}S'.format(interval_sec))
    df.set_index("timestamp", inplace=True)
    volume = df.groupby('timestamp')['size'].sum().rename('volume')
    ohlc = df['price'].resample('1T').ohlc()
    df_merged = ohlc.merge(volume, left_index=True, right_index=True,how="left")
    df_merged = df_merged.fillna(method="ffill")
    df_merged = df_merged.rename(columns={"open":"op","high":"hi","low":"lo","close":"cl"})
    return df_merged
# データの保存場所を指定
data_dir = "data/BTC_JPY/2024/"
interval_sec = 60
df = pd.DataFrame()

for file_path in sorted(glob.glob(data_dir + "*.gz")):
    tmp_df = make_df(file_path,60)
    df = pd.concat([df,tmp_df],axis=0)

こちらのコードを実行すると
変数 df が生成されます。

チュートリアルで用いられている
データを読み込んだ変数になっているので
そのまま使用する事ができると思います。

データが読み込めたかどうかは
次のコードで確認してみて下さい。
df[["cl"]].plot(figsize=(12,4))
plt.show()
download


前処理部分のコードの追加

コードの前処理部分の最初で
手数料を記入するコードがあります。

チュートリアルが作成されてからだいぶ経っており
手数料などが更新されていないため
適宜追加してみて下さい。

maker_fee_history = [
    {
        # https://coin.z.com/jp/news/2020/08/6482/
        # 変更時刻が記載されていないが、定期メンテナンス後と仮定
        'changed_at': '2020/08/05 06:00:00Z',
        'maker_fee': -0.00035
    },
    {
        # https://coin.z.com/jp/news/2020/08/6541/
        'changed_at': '2020/09/09 06:00:00Z',
        'maker_fee': -0.00025
    },
    {
        # https://coin.z.com/jp/news/2020/10/6686/
        'changed_at': '2020/11/04 06:00:00Z',
        'maker_fee': 0.0
    },

    ### 追加
    {
        # 現在値
        'changed_at': '2023/08/05 06:00:00Z',
        'maker_fee': -0.0003
    },
]

あとはチュートリアルを
上から実行していけば
動いていくだろうと思います。

累積リターンのところで
次のような画像が出ればほぼほぼ成功してると思います。

download-1



特徴量の重要度の算出


一般的な機械学習では
予測にどの変数が寄与したのか
重要度を算出することが多いです。

こちらがチュートリアルの
モデルの重要度を算出するコードになります。
feature_importance = model.feature_importances_

# 特徴量名と重要度を紐づける
feature_importance_df = pd.DataFrame({'Feature': features, 'Importance': feature_importance})

# 重要度の降順でソート
feature_importance_df = feature_importance_df.sort_values(by='Importance', ascending=False).reset_index(drop=True)

# 結果の表示(トップ10)
feature_importance_df.head(10)
Feature Importance
0 HT_DCPERIOD 145
1 ADXR 137
2 BBANDS_upperband 125
3 KAMA 124
4 BBANDS_lowerband 116
5 LINEARREG 114
6 MFI 109
7 MACD_macdsignal 107
8 HT_PHASOR_quadrature 106
9 HT_DCPHASE 102


この意味合いとしては
chatGPTによると次の様な意味になっています。



HT_DCPERIOD: Hilbert Transform - Dominant Cycle Period
(ヒルベルト変換 - 優勢サイクル期間)と呼ばれる指標
時系列データの優勢周期を表します。この特徴量は
トレンドのサイクル性を捉えるために使用されます。


ADXR: Average Directional Movement Index Rating
(平均方向運動指数レーティング)は
トレンドの強さを測定する指標

特に、ADX(Average Directional Index)に基づいて計算され
トレンドの強さを示します。


KAMA: Kaufman's Adaptive Moving Average
(カウフマンの適応的移動平均)は移動平均線の一種
価格変動の速さに応じて重みを調整し
より滑らかな移動平均線を提供します。

BBANDS_upperbandおよびBBANDS_lowerband: Bollinger Bands
(ボリンジャーバンド)は、移動平均線を中心に上下に
標準偏差を加えたバンドをプロットする指標

上側バンドと下側バンドは、価格の上下の範囲を示し
価格の過熱やサポート/レジスタンスのレベルを示すのに使用されます。


LINEARREG: Linear Regression(線形回帰)は
価格や指標の値の線形トレンドを推定するための手法です。

この特徴量は、過去のデータからの線形回帰に基づいて
将来のトレンドを予測するために使用されます。


HT_PHASOR_quadratureおよびHT_PHASOR_inphase:
Hilbert Transform - Phasor Components
(ヒルベルト変換 - フェーザーコンポーネント)は
サイン波の位相成分と直交成分を表す指標です。

これらの特徴量はサイクル性や位相関係を捉えるために使用されます。

MFI: Money Flow Index(マネーフローインデックス)は
取引量と価格の関係を用いて買い圧力と売り圧力を計算する指標
主に過買いや過売りの状態を示すのに使用されます。


BETA: ベータは、株式のリスクを示す指標で
市場全体(代表的な指数など)との相関関係を指します。

ベータが1より大きい場合、株価の変動が市場の変動よりも
大きいことを示し、逆にベータが1より小さい場合はその逆を意味します。


ULTOSC: Ultimate Oscillator(アルティメットオシレータ)は
短期、中期、長期の3つの期間を用いてトレンドの強さを計算する指標
主に過買いや過売りの状態を示すのに使用されます。


HT_DCPHASE: Hilbert Transform - Dominant Cycle Phase
(ヒルベルト変換 - 優勢サイクル位相)は
優勢サイクルの位相を表す指標です。この特徴量は
トレンドの位相情報を捉えるために使用されます。


STOCHF_fastk: Stochastic Fast %K
(ストキャスティクスファスト%K)は
価格の変動範囲内での位置を示す指標
主に過買いや過売りの状態を示すのに使用されます。


おまけ

自分もチュートリアルを参考に
別のモデルを構築してみました。

説明変数は同一
目的変数を5分後の価格
に設定し

学習期間は
予測行の前期間step分のみ(step=30)
というモデルにし

1分ずつスライドして
学習しては予測を行うようにしました。

これが予測結果です。

実測と予測値の差分のプロット
download-3

実測と予測値の差分のヒストグラム
download-2

実測と予測のプロット
download-4


学習と予測を近いところで行っているので
ズレすぎることは少ないですね。

最後に予測が現在価格を上まっている条件で
5分後価格と現在価格の差分の累計をプロット
download-5

見事に右肩上がり
予測はかなり機能している様に見えました。


最後に

あくまでシミュレーション上ですが
予測が機能している様に見えるので
売買のロジックに組み込むことも出来るかもしれません。

簡易なロジックとしては
5分後価格予測が現在価格を上まっていたら買い
5分後に売る

5分後価格予測が現在価格を下まっていたら売り
5分後に買う

というような感じですね。

仮の予測モデルは出来ているので
あとは売買ロジック周りを構築すれば
仮想通貨の自動売買botが出来上がります。

その辺りも、もう少し進めていけたらと
思っています。

今回は仮想通貨の自動売買bot本の
チュートリアルを動かしてみたでした。

それでは。

このページのトップヘ