PythonでarXiv.orgを分析してみる

最近の論文のトレンドを調べる方法を探していたのですが、簡単に使えるツールは存在しなさそうでした。

この時、調べたかったことは、次の2点です。

  • 最近、ある分野でよく使われている単語は何か?(ワードクラウドにしてくれると嬉しい)
  • ある単語は、いつ頃使われるようになったのか?(ヒストグラムで見えるようしたい)

一方、科学系論文のプレプリントサーバーarXiv.orgでは、論文の概要を取得するためのarXiv APIが公開されています。

このAPIは、検索クエリを送ると検索結果をAtom形式(RFC4287)のテキストデータを返却するというもので、検索に応じた論文のタイトルやアブストラクトが手に入ります。

そこで、APIでデータを入手して分析するツールを作成してみることにしました。


Arxiv Analyticsツール

作成したのは、以下の2つのクラスです。開発言語はPythonを使っています。

クラス説明
ArxivQueryarXiv APIのクエリを組み立てるクラス。簡易版なので、組み立てるだけで編集はできません。
ArxivData検索結果を格納し、分析するためのクラス。
Arxiv Analyticsツールの内容

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関数で画像ファイルなどに出力できます。

上記の例の場合、次のような画像が出力されました。


まとめ

以上により、目的の分析ができるツールができました。

最近よく使われる単語はArxivData.wordcloud()で見ることができ、使われるようになった時期はArxivData.trend()で見ることができます。

参照文献

  1. arXiv API User’s Manual

ソースコード

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

コメントを残す