Twitterデータのpythonでの収集方法と、時系列のテキストデータに対するバースト検出方法の説明です。
技術的には、以下の過去記事と同様です。
過去記事：
「クッパ姫」に関するツイートをpythonで収集して、バースト検出してみた
https://qiita.com/pocket_kyoto/items/de4b512b8212e53bbba3
この時に採用した方法の汎用性を確認するために、2020年3月10日時点で話題の「コロナ」をキーワードとして、Twitterデータの収集と、「コロナ」と共起する語のバースト検出を実践してみました。
「コロナ」に関するツイートを収集する
収集方法は、基本的に過去記事とほぼ同じです。
まずは、ライブラリの読み込みなど、ツイート収集の準備を行います。
# Twitterデータ収集用のログインキーの情報
KEYS = { # 自分のアカウントで入手したキーを記載
'consumer_key':'*********************',
'consumer_secret':'*********************',
'access_token':'*********************',
'access_secret':'*********************',
}
# Twitterデータの収集（収集準備）
import json
from requests_oauthlib import OAuth1Session
twitter = OAuth1Session(KEYS['consumer_key'],KEYS['consumer_secret'],KEYS['access_token'],KEYS['access_secret'])
Twitterデータ収集用のログインキーの取得方法に関する情報は、参考[1]のサイトが分かりやすいです。
ツイート収集用の関数は、以下のように定義しています。
ツイート場所は今回使わないため、デフォルトの引数（None）を設定できるようにしました。
また、1回あたり最大100ツイートしか検索できないので、for文で繰り返しリクエストする必要があるのですが、Twitterデータ取得関数の外で管理した方がスマートだったので、そのように実装しています。
このあたりは、参考[2]の書き方を踏襲しました。
# Twitterデータ取得関数
def getTwitterData(key_word, latitude=None, longitude=None, radius=None, mid=-1):
url = "https://api.twitter.com/1.1/search/tweets.json"
params ={'q': key_word, 'count':'100', 'result_type':'recent'} #取得パラメータ
if latitude is not None: # latitudeのみで判定
params = {'geocode':'%s,%s,%skm' % (latitude, longitude, radius)}
params['max_id'] = mid # midよりも古いIDのツイートのみを取得する
req = twitter.get(url, params = params)
if req.status_code == 200: #正常通信出来た場合
tweets = json.loads(req.text)['statuses'] #レスポンスからツイート情報を取得
# 最も古いツイートを取るための工夫（※もっと良い書き方ありそう）
user_ids = []
for tweet in tweets:
user_ids.append(int(tweet['id']))
if len(user_ids) > 0:
min_user_id = min(user_ids)
else:
min_user_id = -1
# メタ情報
limit = req.headers['x-rate-limit-remaining'] if 'x-rate-limit-remaining' in req.headers else 0
reset = req.headers['x-rate-limit-reset'] if 'x-rate-limit-reset' in req.headers else 0
return {'tweets':tweets, 'min_user_id':min_user_id, 'limit':limit, 'reset':reset}
else: #正常通信出来なかった場合
print("Failed: %d" % req.status_code)
return {}
上記の関数を連続で実行するための制御関数（getTwitterDataRepeat）を作成しました。
リクエスト制限に引っかからないように、制限に引っかかりそうになったら自動で待機します。
# Twitterデータ連続取得
import datetime, time
def getTwitterDataRepeat(key_word, latitude=None, longitude=None, radius=None, mid=-1, repeat=10):
tweets = []
for i in range(repeat):
res = getTwitterData(key_word, latitude, longitude, radius, mid)
if 'tweets' not in res: #エラーとなった場合は離脱
break
else:
sub_tweets = res['tweets']
for tweet in sub_tweets:
tweets.append(tweet)
if int(res['limit']) == 0: # 回数制限に達した場合は休憩
# 待ち時間の計算. リミット＋５秒後に再開する
now_unix_time = time.mktime(datetime.datetime.now().timetuple()) #現在時刻の取得
diff_sec = int(res['reset']) - now_unix_time
print ("sleep %d sec." % (diff_sec+5))
if diff_sec > 0:
time.sleep(diff_sec + 5)
mid = res['min_user_id'] - 1
print("ツイート取得数：%s" % len(tweets))
return tweets
このように実装することで、リクエストの上限を気にせず、ツイートの収集が自動で可能となります。
あとは、時間帯別でツイートを分けて収集したかったので、以下のようなスクリプトを回しました。
# 参考[3]で作成されていた関数を拝借しました
import time, calendar
def YmdHMS(created_at):
time_utc = time.strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y')
unix_time = calendar.timegm(time_utc)
time_local = time.localtime(unix_time) # 2018/9/24に修正しました
return time.strftime("%Y/%m/%d %H:%M:%S", time_local)
# コロナに関するツイートを6時間おきに1週間分取得する
tweet_corona = {}
mid = -1
for t in range(4*7):
tweets = getTwitterDataRepeat("コロナ", mid=mid, repeat=10)
old_tweet = tweets[-1] # 収集した中で最も古いツイート
key = YmdHMS(old_tweet["created_at"]) # YmdHMS関数
tweet_corona[key] = tweets # 最も古いツイートの時刻をキーとして保存する
mid = old_tweet["id"] - 15099494400000*6 # 約6時間遡って収集する
ツイートを6時間ずつ遡りながら収集したかったので、最も古いツイートのmidから15,099,494,400,000 * 6を引いています。
この15,099,494,400,000という値は、Tweeterのtweet IDの仕様より決めています。
Twitter の tweet ID はミリ秒タイムスタンプ＋ ID を発行するマシンの番号＋シーケンス番号を 64 bit に押し込めた構造になっています。（参考[4]）
「コロナ」に関するツイートを時系列で比較してみる
ここまでで、「コロナ」を含むツイートを時系列で収集することができました。
まずはデータを理解するために、単語の出現頻度を時系列で可視化したいと思います。
以下のような関数を定義して、janomeで形態素解析して、単語の出現頻度をカウントしました。
# 文章を形態素解析して、Bag of Wordsに変換する
from janome.tokenizer import Tokenizer
import collections
import re
def CountWord(tweets):
tweet_list = [tweet["text"] for tweet in tweets]
all_tweet = "\n".join(tweet_list)
t = Tokenizer()
# 原形に変形、名詞のみ、1文字を除去、漢字・平仮名・カタカナの連続飲みに限定
c = collections.Counter(token.base_form for token in t.tokenize(all_tweet)
if token.part_of_speech.startswith('名詞') and len(token.base_form) > 1
and token.base_form.isalpha() and not re.match('^[a-zA-Z]+$', token.base_form))
freq_dict = {}
mc = c.most_common()
for elem in mc:
freq_dict[elem[0]] = elem[1]
return freq_dict
可視化方法はWordCloudを用いました。以下のように実装しました。
# Word Cloudで可視化、WordCloud可視化関数
def color_func(word, font_size, position, orientation, random_state, font_path):
return 'white'
from wordcloud import WordCloud
import matplotlib.pyplot as plt
get_ipython().run_line_magic('matplotlib', 'inline')
from matplotlib.font_manager import FontProperties
fp = FontProperties(fname=r'C:\WINDOWS\Fonts\meiryo.ttc', size=50) #日本語対応
def DrawWordCloud(word_freq_dict, fig_title):
# デフォルト設定を変更して、colormapを"rainbow"に変更
wordcloud = WordCloud(background_color='white', min_font_size=15, font_path='C:\WINDOWS\Fonts\meiryo.ttc',
max_font_size=200, width=1000, height=500, prefer_horizontal=1.0, relative_scaling=0.0, colormap="rainbow")
wordcloud.generate_from_frequencies(word_freq_dict)
plt.figure(figsize=[20,20])
plt.title(fig_title, fontproperties=fp)
plt.imshow(wordcloud,interpolation='bilinear')
plt.axis("off")
これらを用いて、
単語の出現頻度を時系列で可視化します。
datetime_freq_dicts = []
for k, v in sorted(tweet_corona.items()):
print("%s～ のツイート" % k)
freq_dict = CountWord(v)
DrawWordCloud(freq_dict, "%s～ のツイート" % k)
# 単語出現頻度を時系列順に格納
datetime_freq_dicts.append(freq_dict)
出力：
(中略)
:
:
(中略)
:
:
(中略)
:
:
(中略)
:
:
(中略)
:
「新型」「ウイルス」「感染」など、「コロナ」という単語と共起しやすい語群の影響が強く出ました。
この可視化結果からは、「コロナ」の影響で話題になった単語が分かりにくいため、自動検出してみることにします。
コロナの影響で、話題になった単語を自動検出してみる
今回収集したデータセットと、バースト検出と呼ばれる手法を用いて、「コロナ」の影響で、話題になった単語を自動検出してみたいと思います。
バースト検出という手法に関しては、書籍では、「ウェブデータの機械学習 (機械学習プロフェッショナルシリーズ)」に詳しくまとめられているようですが、ネットには解説記事が少ないです。
今回は、自然言語処理に関する研究室として著名な、東北大学 乾・鈴木研究室の解説記事を参考として、バースト検出手法の実装・適用に挑戦してみたいと思います。
今回は、Moving Average Convergence Divergence (MACD) という指標を用いて、バースト検出に取り組んでみました。
バースト検出手法としては、Kleinbergが2002年に発表した手法がベースラインとして、よく用いられるようですが、He and Parker が2010年に発表したMACDのほうがシンプル、かつ計算量が少ないようです。
↓MACDの解説については、乾・鈴木研究室のものが分かりやすいので、そのまま引用したいと思います。
【MACDの解説】
ある時刻におけるMACDは，
MACD = (時系列値の過去f期間の移動指数平均) - (時系列値の過去s期間の移動指数平均)
Signal = (MACD値の過去t期間の移動指数平均)
Histgram = MACD - Signal
ここで，f, s, tはパラメータ（f < s）で，これらをまとめてMACD(f, s, t)と書きます． 今回の実験では，He and Parker (2010) の実験でも用いられていた MACD(4, 8, 5) を採用しました． MACDをテクニカル指標として用いる時は，「Signal < MACD」の状態を上げトレンド，「MACD < Signal」の状態を下げトレンドとし，Histgramがトレンドの強さを表すと言われています． 今回は，15分間の期間をひとまとまりとして（15分足），その期間内にツイッター上に出現した単語の出現頻度を15で割った値，つまり出現速度[回/分]を観測値として，MACDによるトレンド分析を行いました． MACDの計算に必要な移動指数平均の値は逐次計算が可能で，今回のトレンド分析はストリーミング・アルゴリズムとして実装できるため，ビッグデータからのトレンド分析に適していると考えています．
上記の解説内容から、MACDを以下のように実装しました。
# Moving Average Convergence Divergence (MACD) の計算
class MACDData():
def __init__(self,f,s,t):
self.f = f
self.s = s
self.t = t
def calc_macd(self, freq_list):
n = len(freq_list)
self.macd_list = []
self.signal_list = []
self.histgram_list = []
for i in range(n):
if i < self.f:
self.macd_list.append(0)
self.signal_list.append(0)
self.histgram_list.append(0)
else :
macd = sum(freq_list[i-self.f+1:i+1])/len(freq_list[i-self.f+1:i+1]) - sum(freq_list[max(0,i-self.s):i+1])/len(freq_list[max(0,i-self.s):i+1])
self.macd_list.append(macd)
signal = sum(self.macd_list[max(0,i-self.t+1):i+1])/len(self.macd_list[max(0,i-self.t+1):i+1])
self.signal_list.append(signal)
histgram = macd - signal
self.histgram_list.append(histgram)
このプログラムを用いて、2020年3月4日（水）～2020年3月10日（火）にかけて、
コロナの影響で、話題になった単語を自動検出してみたいと思います。
上記の関数へデータを代入するプログラム（折り込み）
# 各時間帯のツイートで、上位100語にランクインする用語をバースト検出する
datetime_freq_dicts = []
for k, v in sorted(tweet_corona.items()):
print("%s～ のツイート" % k)
freq_dict = CountWord(v)
DrawWordCloud(freq_dict, "%s～ のツイート" % k)
# 単語出現頻度を時系列順に格納
datetime_freq_dicts.append(freq_dict)
# 各時間帯のツイートで、上位100語にランクインする用語をバースト検出する
top_100_words = []
i = 0
for freq_dict in datetime_freq_dicts:
for k,v in freq_dict.items():
top_100_words.append(k)
i += 1
if i >= 100:
i = 0
break
top_100_words = list(set(top_100_words)) # ユニークな単語に限定
print(len(top_100_words))
# MACD計算結果の取得
word_list_dict = {}
for freq_dict in datetime_freq_dicts:
for word in top_100_words:
if word not in word_list_dict:
word_list_dict[word] = []
if word in freq_dict:
word_list_dict[word].append(freq_dict[word])
else:
word_list_dict[word].append(0)
# 正規化
word_av_list_dict = {}
for k, v in word_list_dict.items():
word_av_list = [elem/sum(v) for elem in v]
word_av_list_dict[k] = word_av_list
# 計算(He and Parker(2010)と同じパラメータ)
f = 4
s = 8
t = 5
word_macd_dict = {}
for k, v in word_av_list_dict.items():
word_macd_data = MACDData(f,s,t)
word_macd_data.calc_macd(v)
word_macd_dict[k] = word_macd_data
# バースト検出
word_burst_dict = {}
for k,v in word_macd_dict.items():
burst = max(v.histgram_list) # Histgramがトレンドの強さを表すことから、期間内での最大値を取る
word_burst_dict[k] = burst
データを投入した結果は以下の通りです。
i = 1
for k, v in sorted(word_burst_dict.items(), key=lambda x: -x[1]):
print(str(i) + "位：" + str(k))
i += 1
出力：
1位：九郎
2位：ロッテマリーンズ
3位：グラウンド
4位：区役所
5位：尊厳
6位：つば
7位：自習
8位：配達員
9位：メタノール
10位：港北
11位：血清
12位：イープラス
13位：ハラスメント
14位：装置
15位：スナック
16位：佐川急便
17位：リベロ
18位：みゆき
19位：美神
20位：サイケ
21位：ライヴ
22位：横浜市立大学
23位：恐慌
24位：全巻
25位：コロハラ
26位：獣疫
27位：払戻し
28位：登場
29位：義務
30位：表示
:
(中略)
:
「つば」「九郎」「ロッテマリーンズ」などが、「コロナ」の影響で話題になった単語として検出されました。
他の単語についても、概ね納得感のある結果でした。
次いで、話題になった時期の推定も行ってみました。
可視化プログラム（折り込み）
# 結果の可視化
import numpy as np
import matplotlib.pyplot as plt
get_ipython().run_line_magic('matplotlib', 'inline')
from matplotlib.font_manager import FontProperties
fp = FontProperties(fname=r'C:\WINDOWS\Fonts\meiryo.ttc', size=10) #日本語対応
x = np.array(sorted(tweet_corona.keys()))
y1 = np.array(word_macd_dict["ロッテマリーンズ"].histgram_list)
y2 = np.array(word_macd_dict["自習"].histgram_list)
y3 = np.array(word_macd_dict["配達員"].histgram_list)
y4 = np.array(word_macd_dict["メタノール"].histgram_list)
y5 = np.array(word_macd_dict["スナック"].histgram_list)
y6 = np.array(word_macd_dict["ハラスメント"].histgram_list)
plt.plot(x, y1, marker="o")
plt.plot(x, y2, marker="+", markersize=10, markeredgewidth=2)
plt.plot(x, y3, marker="s", linewidth=1)
plt.plot(x, y4, marker="o")
plt.plot(x, y5, marker="+", markersize=10, markeredgewidth=2)
plt.plot(x, y6, marker="s", linewidth=1)
plt.xticks(rotation=90)
plt.title("バースト検出結果", fontproperties=fp)
plt.xlabel("日時", fontproperties=fp)
plt.ylabel("バースト検出結果", fontproperties=fp)
plt.ylim([0,0.2])
plt.legend(["「ロッテマリーンズ」","「自習」", "「配達員」","「メタノール」", "「スナック」","「ハラスメント」"], loc="best", prop=fp)
可視化結果は、以下の通りです。
ヤクルトスワローズ 対 ロッテマリーンズの無観客試合が行われたのが、
3月7日（土）なので、正しく推定できていそうです。
3月10日（火）現在では、「メタノール」が話題の単語の一つとなっているようです。
(3/18追記) 3/11（水）～3/18（水）での結果
3/11（水）～3/18（水）のデータを投入した結果は以下の通りです。
i = 1
for k, v in sorted(word_burst_dict.items(), key=lambda x: -x[1]):
print(str(i) + "位：" + str(k))
i += 1
出力：
1位：規約
2位：サイヤ
3位：マジェスティックレゴン
4位：強靭
5位：民事
6位：地球人
7位：フアン
8位：シティー
9位：大麻
10位：パライソ
11位：闘会議
12位：乱舞
13位：ローラアシュレイ
14位：ミュージカル
15位：不可
16位：推計
17位：蜂蜜
18位：追い打ち
19位：レモン
20位：上演
21位：レシート
22位：刀剣
23位：捜査
24位：マクロン
25位：クラウドファンディング
26位：桶屋
27位：祖母
28位：笑顔
29位：全額
30位：所有
:
(中略)
:
- 五輪チケットは、「規約」上、コロナで中止になっても払い戻し不可、という話題
- ドラゴンボールに登場する「サイヤ」人、悟空の病気はコロナだったのでは？という話題
- コロナにより経営破綻した「マジェスティックレゴン」に関する話題
- 外出自粛により、オランダで「大麻」が大量に購入された話題
- ミュージカル刀剣乱舞~静かなる海「パライソ」~は、イベント開催する話題
- ドワンゴ主催の「闘会議」は、イベント中止する話題
などが瞬間的に話題となったツイートとして検知されました。
話題になった時期は以下の通りです。
まとめと今後
今回は、「コロナ」をテーマにバースト検出に取り組んでみました。
技術的には、過去記事の内容の焼き直しですが、妥当な分析結果が得られたのではないかと考えています。
過去記事では、「クッパ姫」をテーマとしていましたが、手法自体は汎用性が高いことを確認できました。
引き続き、Twitterデータ分析にチャレンジしていきたいと思います。
参考
[1]
【2019年】TwitterのAPIに登録し、アクセスキー・トークンを取得する具体的な方法
https://miyastyle.net/twitter-api
[2]
スタバのTwitterデータをpythonで大量に取得し、データ分析を試みる その１
https://qiita.com/kenmatsu4/items/23768cbe32fe381d54a2
[3]
Streaming APIで取得したつぶやきの処理方法
http://blog.unfindable.net/archives/4302
[4]
スケーラブルな採番とsnowflake
https://kyrt.in/2014/06/08/snowflake_c.html
[5]
東北大学 乾・鈴木研究室 Project 311 / Trend Analysis
http://www.cl.ecei.tohoku.ac.jp/index.php?Project%20311%2FTrend%20Analysis
[6]
Dan He and D. Stott Parker(2010)
「Topic Dynamics: An Alternative Model of 'Bursts' in Streams of Topics」
https://dollar.biz.uiowa.edu/~street/HeParker10.pdf