最近の論文のトレンドを調べる方法を探していたのですが、簡単に使えるツールは存在しなさそうでした。
この時、調べたかったことは、次の2点です。
- 最近、ある分野でよく使われている単語は何か?(ワードクラウドにしてくれると嬉しい)
- ある単語は、いつ頃使われるようになったのか?(ヒストグラムで見えるようしたい)
一方、科学系論文のプレプリントサーバーarXiv.orgでは、論文の概要を取得するためのarXiv APIが公開されています。
このAPIは、検索クエリを送ると検索結果をAtom形式(RFC4287)のテキストデータを返却するというもので、検索に応じた論文のタイトルやアブストラクトが手に入ります。
そこで、APIでデータを入手して分析するツールを作成してみることにしました。
目次
Arxiv Analyticsツール
作成したのは、以下の2つのクラスです。開発言語はPythonを使っています。
クラス | 説明 |
---|---|
ArxivQuery | arXiv APIのクエリを組み立てるクラス。簡易版なので、組み立てるだけで編集はできません。 |
ArxivData | 検索結果を格納し、分析するためのクラス。 |
ArxivQueryクラスは、毎回間違わずにクエリを組み立てるのが面倒だったため作りました。ArxivDataクラスは、Atom形式のデータを入手し、構文解析も終わった状態でデータを保持しています。
ArxivQueryクラス
ArxivQueryクラスを使ったクエリは、次のような記述で作成します。
import arxiv_analytics as ax
query = ax.ArxivQuery()
query.category("cs.LG").AND().abstract("deep learning")
id以外の検索フィールド(category, commentなど)にキーワードを設定し、論理演算子AND, OR, ANDNOTのいずれかで連結していきます。もし、論理演算子が指定されていなければ、ANDが指定されたものとしてクエリが作られます。queryは、__str__関数が呼ばれたときに、クエリ文字列に変換されます。
上記の記述の場合のクエリは、次のようになります。
http://export.arxiv.org/api/query?search_query=cat:cs.LG+AND+abs:%22deep+learning%22&start=0&max_results=10
ArxivQueryクラスに定義されているメンバー関数は、次の通りです。
関数 | 説明 |
---|---|
ArxivQuery(start=0, max_results=10) | コンストラクタ |
clear() | 構築中のクエリ消去 |
start( n ) | 検索結果の中から、取り出す結果の開始番号(default=0) |
max_results( n ) | 最大検索数(default=10) |
sortby( text ) | 検索結果のソート方法。”relevance”(妥当順、規定値)、”lastUpdatedDate”(更新日順)、”submittedDate”(投稿日順)から選択する。 |
sortorder( text ) | 検索結果のソート順序。”ascending”(昇順、規定値)、”descending”(降順)から選択する。 |
title( query ) | 題名 |
author( query ) | 著者 |
abstract( query ) | 概要 |
comment( query ) | コメント |
journal( query ) | 掲載された論文誌 |
category( query ) | カテゴリ(例:hep-th、cs.LGなど) |
id( query ) | 文献のID番号(例:2204.00213など) |
report_number( query ) | 文献のID番号(例:2204.00213など) |
all( query ) | 全フィールド |
AND() | 論理演算ANDを追加 |
OR() | 論理演算ORを追加 |
ANDNOT() | 論理演算ANDNOTを追加 |
open_parens() | “(“を追加 |
close_parens() | “)”を追加 |
ArxivDataクラス
ArxivDataクラスは、クエリに基づいてarXiv.orgのデータを取得し、構文解析をして操作可能なデータに変換します。
import arxiv_analytics as ax
query = ax.ArxivQuery()
query.category("cs.LG").AND().abstract("deep learning")
data = ax.ArxivData(query)
queryは、ArxivQueryオブジェクトを与えていますが、正しいクエリであれば文字列でも構いません。
ArxivDataクラスに定義されているメンバー関数は、次の通りです。
関数 | 説明 |
---|---|
ArxivData(query) | コンストラクタ |
reload( query ) | クエリを変更して、再度データを取得する。 |
query() | クエリオブジェクト(またはテキスト)を返す。 |
text() | 取得したテキストデータを返す。 |
parsed() | 構文解析済みの辞書オブジェクトを返す。 |
collect( tag ) | 検索でヒットした複数の結果から、tag要素だけを取り出しリスト型で返す。例えば、題名リストはtag=title、概要リストはtag=summaryとする。 |
histgram( tag ) | 検索でヒットした複数の結果から、tag要素の出現頻度(ヒストグラム)を作成し、pandas.DataFrameオブジェクトで返却する。 |
wordcloud( tag, background_color=”white”, width=800, height=600, stop_words=None) | 検索でヒットした複数の結果から、tag要素だけを対象としたワードクラウドを作成する。結果は、wordcloud.WordCloudオブジェクトで返却する。 |
trend() | 検索でヒットした複数の結果から、出版年月日についてのヒストグラムデータを作成する。結果は、pandas.DataFrameオブジェクトで返却する。 |
現在は、分析手法は、collect, wordcloud, trendの3つしかありません。今後、気が向けは追加するかもしれません。
Arxiv Analyticsの結果例
上記のツールを使って、実際に分析した結果を例示したいと思います。
分析には次のようなコードを用いました。
import pandas as pd
import wordcloud as wc
import arxiv_analytics as ax
query = ax.ArxivQuery()
query.max_results(1000)
query.sortby("submittedDate")
query.sortorder("descending")
query.category("cs.LG").AND().abstract("deep learning")
print(query)
data = ax.ArxivData(query)
collection = data.collect('title')
print(collection)
trends = data.trend()
print(trends)
wordcloud = data.wordcloud('summary')
wordcloud.to_file("wordcloud.png")
クエリは、コンピュータサイエンスの機械学習カテゴリ(cs.LG)の中で、概要で深層学習に触れている論文を検索するものです。ただし、投稿日順に新しい方(降順)から1,000件までとしています。
collect関数の結果
collect関数は、指定された要素の一覧を取得し、リスト型で返却します。
結果は、以下のような、題名の一覧を取得しています。
['TorchSparse: Efficient Point Cloud Inference Engine', 'Deep learning techniques for energy clustering in the CMS ECAL', 'Learning spatiotemporal features from incomplete data for traffic flow prediction using hybrid deep neural networks', 'OCTOPUS -- optical coherence tomography plaque and stent analysis software', 'Automated analysis of fibrous cap in intravascular optical coherence tomography images of coronary arteries', 'On Distribution Shift in Learning-based Bug Detectors', 'A data filling methodology for time series based on CNN and (Bi)LSTM neural networks', 'Assessing Machine Learning Algorithms for Near-Real Time Bus Ridership Prediction During Extreme Weather', 'Estimating city-wide hourly bicycle flow using a hybrid LSTM MDN', 'Deep Reinforcement Learning for a Two-Echelon Supply Chain with Seasonal Demand', 'Hephaestus: A large scale multitask dataset towards InSAR understanding', 'A Probabilistic Time-Evolving Approach to Scanpath Prediction', 'Ordinal-ResLogit: Interpretable Deep Residual Neural Networks for Ordered Choices', 'Robustness Testing of Data and Knowledge Driven Anomaly Detection in Cyber-Physical Systems', 'Restructuring TCAD System: Teaching Traditional TCAD New Tricks', ....]
trend関数の結果
trend関数は、投稿年月日を調べて、ヒストグラムデータを作成します。戻り値は、pandas.DataFrame型なので、グラフに出力するなどの加工は自由に行えます。
上記の例の場合、次のような結果が表示されます。
number of papers
2022-01 242
2022-02 305
2022-03 304
2022-04 149
wordcloud関数の結果
wordcloud関数は、wordcloudパッケージのWordCloudオブジェクトを返却します。そのため、WordCloudクラスのto_file関数で画像ファイルなどに出力できます。
上記の例の場合、次のような画像が出力されました。
![](https://i0.wp.com/knowledge-bridge.info/wp-content/uploads/2022/04/wordcloud-1.png?resize=800%2C600&ssl=1)
まとめ
以上により、目的の分析ができるツールができました。
最近よく使われる単語はArxivData.wordcloud()で見ることができ、使われるようになった時期はArxivData.trend()で見ることができます。
参照文献
ソースコード
GitHub:https://github.com/jyam45/arxiv_analytics src/arxiv_analytics.py
import feedparser as fp
import datetime as dt
import pandas as pd
import wordcloud as wc
# arXiv API User's Manual
#
# https://arxiv.org/help/api/user-manual
#
class ArxivQuery :
def __init__(self,start=0,max_results=10):
self.API_URL_ = 'http://export.arxiv.org/api/query'
self.query_list_ = []
self.id_list_ = []
self.start_ = start
self.max_results_ = max_results
self.sortby_ = None # { "relevance" | "lastUpdatedDate" | "submittedDate" }
self.sortorder_ = None # { "ascending" | "descending" }
def clear( self ):
self.query_list_.clear()
self.id_list_.clear()
self.start_ = 0
self.max_results_ = 10
self.sortby_ = None
self.sortorder_ = None
def start( self, n ):
if n >= 0 :
self.start_ = n
else:
raise ValueError("Illegal value.")
return self
def max_results( self, n ):
if n > 0 and n < 30001:
self.max_results_ = n
else:
raise ValueError("Illegal value.")
return self
def sortby( self, text ):
items = [ "relevance" , "lastUpdatedDate", "submittedDate" ]
if text in items:
self.sortby_ = text
else:
raise ValueError("Illegal value.")
return self
def sortorder( self, text ):
items = [ "ascending", "descending" ]
if text in items:
self.sortorder_ = text
else:
raise ValueError("Illegal value.")
return self
@staticmethod
def _escape(text):
if ' ' not in text: return text
query = "%22" + text.replace(' ','+') + "%22"
return query
def title( self, query ):
self.query_list_.append("ti:"+self._escape(query))
return self
def author( self, query ):
self.qeury_list_.append("au:"+self._escape(query))
return self
def abstract( self, query ):
self.query_list_.append("abs:"+self._escape(query))
return self
def comment( self, query ):
self.query_list_.append("co:"+self._escape(query))
return self
def journal( self, query ):
self.query_list_.append("jr:"+self._escape(query))
return self
def category( self, query ):
self.query_list_.append("cat:"+self._escape(query))
return self
def id( self, query ):
self.id_list_.append(self._escape(query))
return self
def report_number( self, query ):
self.query_list_.append("rn:"+self._escape(query))
return self
def all( self, query ):
self.query_list_.append("all:"+self._escape(query))
return self
def AND( self ):
self.query_list_.append("AND")
return self
def OR( self ):
self.query_list_.append("OR")
return self
def ANDNOT( self ):
self.query_list_.append("ANDNOT")
return self
def open_parens( self ):
self.query_list_.append("%28")
return self
def close_parens( self ):
self.query_list_.append("%29")
return self
def _make_search_query( self ):
boolean=["AND", "OR", "ANDNOT"]
needs_boolean=False
search_query = ""
for query in self.query_list_:
if query not in boolean:
if needs_boolean :
search_query += "+AND+"
search_query += query
needs_boolean = True
elif query == "(":
if not needs_boolean:
search_query += query
else:
raise ValueError("Query's syntax error.")
elif query == ")":
if needs_boolean:
search_query += query
else:
raise ValueError("Query's syntax error.")
else:
if needs_boolean:
search_query += "+" + query + "+"
needs_boolean = False
else:
raise ValueError("Query's syntax error.")
if not needs_boolean :
raise ValueError
return search_query
def _make_id_list(self):
return ' '.join(self.id_list_)
def __str__(self):
out_list = []
# search_queryの文字列化
search_query = self._make_search_query()
if search_query != "":
out_list.append( "search_query="+search_query )
# id_listの文字列化
id_list = self._make_id_list()
if id_list != "":
out_list.append( "id_list="+'"'+search_query+'"' )
# start
out_list.append("start="+str(self.start_))
# max_results
out_list.append("max_results="+str(self.max_results_))
# sortBy
if self.sortby_ is not None:
out_list.append("sortBy="+self.sortby_)
# sortOrder
if self.sortorder_ is not None:
out_list.append("sortOrder="+self.sortorder_)
query = self.API_URL_ + "?" + '&'.join(out_list)
return query
class ArxivData:
def __init__( self, query ):
self.stop_list_ = ["a","the","it","is","are","was","were","not","do","did","no","any","there","i","we","to","in","of","for","with","this","that"]
self.reload(query)
def reload( self, query ):
self.query_ = query
self.raw_data_ = urllib.request.urlopen(str(query))
self.utf_data_ = self.raw_data_.read().decode('utf-8')
self.obj_data_ = fp.parse(self.utf_data_)
def __str__(self):
return str(self.obj_data_)
def query(self):
return self.query_
def text(self):
return self.utf_data_
def parsed(self):
return self.obj_data_
def collect(self,tag):
items=[]
for entry in self.obj_data_['entries']:
item = entry[tag].replace('\n','').replace(' ',' ')
items.append(item)
return items
def histgram(self,tag):
hist={}
for entry in self.obj_data_['entries']:
item = entry[tag].replace('\n','').replace(' ',' ')
if not item in hist: hist[item]=0
hist[item]+=1
return pd.DataFrame.from_dict(hist, orient="index", columns=["count"])
def wordcloud(self,tag,background_color="white",width=800,height=600,stop_words=None):
# 分析対象の要素から文章を抽出し、1つの文字列にまとめる
text=""
for entry in self.obj_data_['entries']:
text += entry[tag].replace('\n','').replace(' ',' ').replace('.','').replace(',','').lower()
if text == "":
raise ValueError("No entry.")
# 文字列を空白で分割し、意味のない単語リストstop_wordsに登録された単語を削除し、再度連結して1つの文字列にする
word_list = text.split(' ')
stop_list = stop_words if stop_words is not None else self.stop_list_
words = [word for word in word_list if word not in stop_list ]
text = ' '.join(words)
# WordCloudオブジェクトを返却する
return wc.WordCloud(background_color=background_color,width=width,height=height).generate(text)
def trend(self):
# 出版日時をdatetimeオブジェクトに変換し、日単位の検索期間を計算する
pub_dates = self.collect('published')
dt_list = []
for pub_date in pub_dates:
dt_obj = dt.datetime.strptime(pub_date,'%Y-%m-%dT%H:%M:%SZ')
dt_list.append(dt_obj)
if not dt_list :
raise ValueError("No entry.")
period_days = (max(dt_list) - min(dt_list)).days
# 検索期間に応じて、集計単位を変更する。2ヶ月以下なら日単位、2年以下なら月単位、それ以上は年単位
if period_days < 61:
key_format = "%Y-%m-%d"
dt_stride = dt.timedelta(days=1)
elif period_days < 731:
key_format = "%Y-%m"
dt_stride = dt.timedelta(days=30)
else:
key_format = "%Y"
dt_stride = dt.timedelta(days=365)
# 日付をキーにヒストグラムを作成する。
hist={}
dt_itr = min(dt_list)
dt_end = max(dt_list)
while (dt_end - dt_itr).days >= 0:
key = dt.datetime.strftime(dt_itr,key_format)
hist[key]=0
dt_itr = dt_itr + dt_stride
for dt_obj in dt_list:
key = dt.datetime.strftime(dt_obj,key_format)
if not key in hist : hist[key]=0
hist[key]+=1
# pandas.DataFrameオブジェクトで返却する
return pd.DataFrame.from_dict(hist, orient="index", columns=["number of papers"])
arxiv_analytics.pyをsample.pyと同じディレクトリに入れて、以下のコマンドを実行してください。
$ python3 sample.py