仮想サーファーの波乗り

仮想化エンジニアの日常

プログラミング・SNS分析・仮想通貨・自動化などに関してよく書く雑記ブログ

仮想通貨データをCoinGeckoのAPIで取得してPythonで遊ぶ② ~仮想通貨価格を知らせるTwitterBot開発~


今回は、CoinGeckoのAPIにアクセスして取得したデータをもとに、つぶやきをしてくれるTwitterのBotを開発していきます。何もしなくても自動でリアルタイム情報をつぶやいてくれるので、仮想通貨の価格どうなったかな?って時に活用できて便利かなと。

Pythonの実行環境とHerokuにデプロイできる環境はすでにあることを前提に進めていきます。


開発したTwitter Botはどんなもの?

完成したBotがこちら!

予想外に実装に手こずり、現在の機能を実装しきるまでに1日程度かかってしまいました...。

機能要件としては以下のようなものにしています。

①主要コインの現在価格をツイート(1時間に1回)
②主要アルトコインの現在価格をツイート(1時間に1回)
③個人的に好きなコインの現在価格をツイート(1時間に1回)
④有名メディア/ブロガーの直近1時間で公開された記事をRSSで取得してきてツイート(1時間に1回)
⑤直近24時間での値動きのうち上昇率/下落率のTOP3つをツイート(1時間に1回)

定期的にチャートなどで確認するデータをTwitterで知れたらいいなと思って実装しましたが、特に「直近24時間での値動きのうち上昇率/下落率のTOP3つをツイート」は上昇率・下落率の大きくなった時期には面白いツイートになるんじゃないかと思ってます。

2017年の12月に草コインが1日で5倍になったりしていた時期が懐かしいです。


Twitter Botの開発方法

簡単に今回開発したTwitter Botの開発方法をまとめておきます。


Twitter Botアプリケーションの開発/デプロイ環境構築

まずは、Pythonの環境構築とHerokuへのアプリケーションのデプロイをして、アプリケーションを開発する準備をします(この段階はこちらの開発時に終わっていたので、特にやることはありませんでした)。

アプリケーション開発の準備ができたら、Twitterのアカウントを作成し、Twitter APIのAPIキーを取得しておきます。


Twitter Botアプリケーションの開発

環境構築ができたら、具体的な実装に入っていきます。

以下のように、Botの機能を実装したファイルを作成していきます。


coingecko_service.py

主にCoinGecko APIへのアクセス、取得したデータの加工などの役割をもたせたファイル。

# coding=utf-8
import requests
import json
from datetime import datetime

api_url = 'https://api.coingecko.com/api/v3/coins/'


def create_crypt_info_tweet_text(ids):
    current = datetime.now()
    tweet_text = '【{}年{}月{}日 {}時{}分の値動き】'.format(current.year, current.month, current.day, current.hour, current.minute)
    for id in ids:
        tweet_text = tweet_text + '\n' + create_crypt_info_text(id)
    return tweet_text


def create_crypt_info_text(id):
    symbol = convert_id_to_symbol(id)
    json_dict = request_coin_data(id).json()
    price = create_crypt_shortened_price(json_dict)
    price_change = create_shortened_price_change_percentage_24h(json_dict)
    return '{} ¥{}(24h:{}%)'.format(symbol, price, price_change)


def request_coins_data():
    apiPath = api_url
    payload = {'Content-Type': 'application/json', 'order': 'gecko_desc', 'per_page': '200', 'page': 1}
    response = requests.get(apiPath, params=payload)
    return response


def request_coin_data(id):
    apiPath = api_url + id
    payload = {'Content-Type': 'application/json'}
    response = requests.get(apiPath, params=payload)
    return response


def create_crypt_shortened_price(json_dict):
    price_str = extract_jpy_price(json_dict)
    return str(price_str)[:7]


def create_shortened_price_change_percentage_24h(json_dict):
    price_change = extract_price_change_percentage_24h(json_dict)
    return shorten_price_change_percentage(price_change)


def shorten_price_change_percentage(price_change):
    price_change_str = str(price_change[:5])
    if price_change_str[-1:] == ".":
        return price_change_str[:-1]
    else:
        return price_change_str


def create_price_change_text(extract_num):
    text = "【24時間の値動きランキング】"
    json_dict = request_coins_data().json()
    text = '{}{}価格上昇⤴︎TOP3 売り時😌?'.format(text, '\n\n')
    price_change_percent_desc_dict = extract_price_change_percent_desc(json_dict, extract_num, True)
    rank_num = 1
    for name, percent in price_change_percent_desc_dict.items():
        percent_str = shorten_price_change_percentage(percent)
        text = '{}{}第{}位 {} {}%'.format(text, '\n', str(rank_num), name, percent_str)
        rank_num = rank_num + 1
    text = '{}{}価格下落⤵︎WORST3 買い時😇?'.format(text, '\n\n')
    price_change_percent_asc_dict = extract_price_change_percent_desc(json_dict, extract_num, False)
    rank_num = 1
    for name, percent in price_change_percent_asc_dict.items():
        percent_str = shorten_price_change_percentage(percent)
        text = '{}{}第{}位 {} {}%'.format(text, '\n', str(rank_num), name, percent_str)
        rank_num = rank_num + 1
    return text


def extract_jpy_price(json_dict):
    return json_dict['market_data']['current_price']['jpy']


def extract_price_change_percentage_24h(json_dict):
    return json_dict['market_data']['price_change_percentage_24h']


def extract_price_change_percent_desc(json_dict, extract_num, is_desc):
    # json_dictからnameと変動のdictに入れ替える
    price_change_dict = {}
    price_change_dict_for_return = {}
    for json in json_dict:
        name = json['name']
        price_change_24h = json['market_data']['price_change_percentage_24h']
        price_change_dict[name] = price_change_24h
    if is_desc:
        # 価格上昇率の大きい順にnameと変動値をならべかえる
        # sort一発でしたいけど、strを持っているdictionaryをfloat型に変換してsortする方法がわからずvalueで検索しなおしている。
        loop_count = 0
        for price_change in sorted(price_change_dict.values(), key=float, reverse=True):
            loop_count = loop_count + 1
            if loop_count > extract_num:
                return price_change_dict_for_return
            for k, v in price_change_dict.items():
                if v == price_change:
                    price_change_dict_for_return[k] = price_change
    else:
        # 価格下落率の大きい順にnameと変動値をならべかえる
        loop_count = 0
        for price_change in sorted(price_change_dict.values(), key=float):
            loop_count = loop_count + 1
            if loop_count > extract_num:
                return price_change_dict_for_return
            for k, v in price_change_dict.items():
                if v == price_change:
                    price_change_dict_for_return[k] = price_change
    return price_change_dict_for_return


def convert_id_to_symbol(id):
    if id == 'bitcoin':
        return 'BTC'
    elif id == 'ethereum':
        return 'ETH'
    elif id == 'ripple':
        return 'XRP'
    elif id == 'bitcoin-cash':
        return 'BCH'
    elif id == 'nem':
        return 'XEM'
    elif id == 'neo':
        return 'NEO'
    elif id == 'lisk':
        return 'LSK'
    elif id == 'stellar':
        return 'XLM'
    elif id == 'monacoin':
        return 'MONA'
    elif id == 'verge':
        return 'XVG'
    elif id == 'omisego':
        return 'OMG'
    elif id == 'tron':
        return 'TRX'
    elif id == 'storj':
        return 'STORJ'
    elif id == 'status':
        return 'SNT'
    elif id == 'quantstamp':
        return 'QSP'


tweet_service.py

主にTwitterへの投稿機能を実装したファイル。

# coding=utf-8
import tweepy
import os


# ツイートをする
def tweet_text(screen_name, text):
    api =prepare_twitter_api(screen_name)
    api.update_status(text)


# 複数ツイートをする
def tweet_multiple_times(text_list):
    api =prepare_twitter_api(screen_name)
    print(text_list)
    for text in text_list:
        print(text)
        api.update_status(status=text)


# アカウントのTwitterAPIを用意
def prepare_twitter_api(user_screen_name):
    auth = tweepy.OAuthHandler(os.environ['CRYPT_CONSUMER_KEY'], os.environ['CRYPT_CONSUMER_SECRET'])
    auth.set_access_token(os.environ['CRYPT_ACCESS_TOKEN'], os.environ['CRYPT_ACCESS_TOKEN_SECRET'])
    return tweepy.API(auth)


rss_service.py

RSS経由で対象としているブログ・メディアが記事を更新したらその記事のリンクを取得する役割をもたせている。

import feedparser
from datetime import datetime


# 指定時間以内に公開された記事のリンクとdescription一覧を元にツイート文章生成
def create_rss_tweet_text_list(rss_list, hour):
    tweet_text_list = []
    entry_dict = extract_article_dict_with_rss(rss_list, hour)
    print(entry_dict)
    # dictionaryのvalueにNoneとstrが含まれているとforで取得できないっぽいのでkeyだけ取得していく
    for link in entry_dict:
        if entry_dict[link] is None:
            tweet_text_list.append(link)
        else:
            tweet_text = '> {}\n\n{}'.format(entry_dict[link], link)
            tweet_text_list.append(tweet_text)
    return tweet_text_list


# 指定時間以内に公開された記事のリンクとdescription一覧を取得する
def extract_article_dict_with_rss(rss_list, hour):
    current_datetime = datetime.now()
    target_entry_dict = {}
    for rss_url in rss_list:
        rss_dict = feedparser.parse(rss_url)
        for entry in rss_dict.entries:
            published_datetime = entry.published.split(' ')
            published_month = convert_month_str_int(published_datetime[2])  # May, Apr, ... -> 3, 4, ...
            if published_datetime[1][1:] == '0':  # 00~31 -> 0~31
                published_day = int(published_datetime[1][:1])
            else:
                published_day = int(published_datetime[1])
            published_time = published_datetime[4]
            split_time = published_time.split(':')
            # UTC時間が0~9の場合、9~18にする
            if split_time[0][:1] == '0':
                published_hour = int(split_time[0][1:]) + 9
            # UTC時間が10~14の場合、19~23にする
            elif int(split_time[0][:1]) <= 14:
                published_hour = int(split_time[0]) + 9
            # UTC時間が15~23の場合、0~8にする
            else:
                published_hour = int(split_time[0]) - 15
            if split_time[1][:1] == '0':
                published_minute = int(split_time[1][1:])
            else:
                published_minute = int(split_time[1])
            if current_datetime.month == published_month and current_datetime.day == published_day \
                    and ((current_datetime.hour * 60) + current_datetime.minute) - ((published_hour * 60) + published_minute) <= hour * 60:
                # descriptionがplainの場合はmax100文字として取得(ツイート文字数制限を考慮)
                if '</' not in entry.description:
                    target_entry_dict[entry.link] = entry.description[:100]
                # descriptionがhtml形式の場合はNoneで渡す
                else:
                    target_entry_dict[entry.link] = None
    return target_entry_dict


def convert_month_str_int(str):
    if str == 'Jan':
        return 1
    elif str == 'Feb':
        return 2
    elif str == 'Mar':
        return 3
    elif str == 'Apr':
        return 4
    elif str == 'May':
        return 5
    elif str == 'Jun':
        return 6
    elif str == 'Jul':
        return 7
    elif str == 'Aug':
        return 8
    elif str == 'Sep':
        return 9
    elif str == 'Oct':
        return 10
    elif str == 'Nov':
        return 11
    elif str == 'Dec':
        return 12


batch.py

10分ごと定期的に呼び出されるバッチファイル。これをHerokuのSchedulerで10分ごとに起動することで、特定の時間にBotがツイートをしてくれます。

# coding=utf-8
import twitter_service
import coingecko_service
import rss_service


# 記事のリンクを取得する対象のメディア・ブログ一覧
REINA_RSS_URL = "https://bitcoin-yoro.com/feed"
OTAKU_RSS_URL = "https://coinotaku.com/feed"
ME_RSS_URL = "https://isamist.work/feed"
SEIYA_RSS_URL = "https://seiyablog.com/feed"
COINDESK_RSS_URL = "https://feeds.feedburner.com/CoinDesk"
COINPOST_RSS_URL = "http://coinpost.jp/?feed=rss2"
BTCNEWS_RSS_URL = "https://btcnews.jp/feed/"
BITTIMES_RSS_URL = "https://bittimes.net/tag/crypto-currency/feed"
CRYPTTIMES_RSS_URL = "https://crypto-times.jp/feed/"
MANA_RSS_URL = "https://bitcoiner.link/feed"
NILS_RSS_URL = "https://altcoins.blue/feed/"
HIGASHI_RSS_URL = "http://coinandpeace.hatenablog.com/rss"
GUNOSY_RSS_URL = "http://blockchain.gunosy.io/rss"
GINCO_RSS_URL = "https://magazine.ginco.io/index.xml"
RSS_LIST = [REINA_RSS_URL, OTAKU_RSS_URL, ME_RSS_URL, SEIYA_RSS_URL, COINDESK_RSS_URL, COINPOST_RSS_URL, BTCNEWS_RSS_URL,
            BITTIMES_RSS_URL, CRYPTTIMES_RSS_URL, MANA_RSS_URL, NILS_RSS_URL, HIGASHI_RSS_URL, GUNOSY_RSS_URL, GINCO_RSS_URL]


def execute():
    current_hour = datetime.current_datetime().hour
    # 2時~8時はおやすみ(処理しない)
    if 2 < current_hour < 8:
        return

    current_minute = datetime.current_datetime().minute

    # 直近24時間での値動き(値上がり/値下がり大きい順に3つ表示)
    if 0 <= current_minute < 10:
        text = coingecko_service.create_price_change_text(3)
        twitter_service.tweet_text(user_screen_name, text)

    # 20分に1回それぞれ違う種類の仮想通貨価格をツイート
    if 0 <= current_minute < 10:
        text = coingecko_service.create_crypt_info_tweet_text(['bitcoin', 'ethereum', 'ripple', 'bitcoin-cash', 'nem'])
        twitter_service.tweet_text(user_screen_name, text)
    if 20 <= current_minute < 30:
        text = coingecko_service.create_crypt_info_tweet_text(['neo', 'lisk', 'stellar', 'monacoin', 'verge'])
        twitter_service.tweet_text(user_screen_name, text)
    if 40 <= current_minute < 50:
        text = coingecko_service.create_crypt_info_tweet_text(['omisego', 'tron', 'storj', 'status', 'quantstamp'])
        twitter_service.tweet_text(user_screen_name, text)

    # 直近1時間での有名メディアでの記事更新情報つぶやき
    if 30 <= current_minute < 40:
        text_list = rss_service.create_rss_tweet_text_list(RSS_LIST, 1)
        if text_list is not None:
            twitter_service.tweet_multiple_times(user_screen_name, text_list)


ファイルの作成ができたら、環境変数にAPIキーの値を設定(.envファイルを修正)します。

.env

# Twitter API(仮想通貨ニュース @cryptNewsBot)
CRYPT_ACCESS_TOKEN=xxx
CRYPT_ACCESS_TOKEN_SECRET=xxx
CRYPT_CONSUMER_KEY=xxx
CRYPT_CONSUMER_SECRET=xxx


あとは、Local環境でbatch.pyファイルを実行して動作確認をするために必要な各種module(各ファイルでimportしているmodule)をpipでインストールしていきます。そして、Heroku上でも同様のmoduleを利用できるようにするために、requirements.txtにmodule名とバージョンを指定しておきます。

$ pip install {必要なモジュール}
$ pip freeze > requirement.txt


最後にherokuにデプロイして、Heroku Schedulerを設定しておけばアプリケーションの実装は完了です。


まとめ

ツイッターBotを実装しようと思い立って ↑ このツイートをしてから、ほぼ1日で実装することができました。

TwitterのBot開発は自分の欲しい情報を通知すれば情報収集の効率化になるし、自分の作りたい機能を最速で実装するためにプログラミングの効果的な学習ができるし、一度開発してしまえば他の分野でのBotも簡単に開発できる(プログラム使いまわせる)から次実装するときは楽に開発できるし、もしかしたらお小遣い稼ぎくらいはできるかも(会社の上司は70個のアカウントを自動で運用して月に70万稼いでいた時期があったそう。そこまでいくともはや本業並みの副業...)などなど、開発しててメリットは多いので、興味ある人はやってみると面白いかと!

半年前にBot開発未経験の頃は、「TwitterのBotとか作れたら楽でいいだろうな〜。でも作るの大変そうだし、簡単には無理だろうな〜。」とか思っていました。ですが、仕事後の夜の空いた時間などを使って、ちょこちょこ調べながら実装進めていけば全くわからない状態から2週間程度で実装でき、今回は1日で実装できました!

1回目はやる前に難しいと思っていたよりは簡単で、2回目以降は生産性が10倍以上になっていくエンジニアリングあるある。


とりあえず今回のBotを育成して有意義な情報をつぶやいてくれるBotにしていきます!


では!