Analise dados de séries temporais com Python e MongoDB usando PyMongoArrow e Pandas

1
Jupyter Notebook

1
import pymongo
2
import os
3
4
from pymongo import MongoClient
5
6
MONGO_CONN_STRING = os.environ.get("MONGODB_CONNECTION_STRING")
7
8
client = MongoClient(MONGO_CONN_STRING)

1
# Let's create a new database called "stock data"
2
db = client.stock_data
3
4
# Let's create a new time-series collection in the "stock data" database called "stocks"
5
6
collection = db.create_collection('stocks', timeseries={
7
8
          timeField: "timestamp",
9
          metaField: "metadata",
10
          granularity: "hours"
11
12
})

1
# Create some sample data
2
3
data = [
4
{
5
   "metadata": {
6
      "stockSymbol": "ABC",
7
      "exchange": "NASDAQ"
8
   },
9
   "timestamp": datetime(2023, 9, 12, 15, 19, 48),
10
   "open": 54.80,
11
   "high": 59.20,
12
   "low": 52.60,
13
   "close": 53.50,
14
   "volume": 18000
15
},
16
17
{
18
   "metadata": {
19
      "stockSymbol": "ABC",
20
      "exchange": "NASDAQ"
21
   },
22
   "timestamp": datetime(2023, 9, 12, 16, 19, 48),
23
   "open": 51.00,
24
   "high": 54.30,
25
   "low": 50.50,
26
   "close": 51.80,
27
   "volume": 12000
28
},
29
30
{
31
   "metadata": {
32
      "stockSymbol": "ABC",
33
      "exchange": "NASDAQ"
34
   },
35
   "timestamp":datetime(2023, 9, 12, 17, 19, 48),
36
   "open": 52.00,
37
   "high": 53.10,
38
   "low": 50.50,
39
   "close": 52.90,
40
   "volume": 10000
41
},
42
43
{
44
   "metadata": {
45
      "stockSymbol": "ABC",
46
      "exchange": "NASDAQ"
47
   },
48
   "timestamp":datetime(2023, 9, 12, 18, 19, 48),
49
   "open": 52.80,
50
   "high": 60.20,
51
   "low": 52.60,
52
   "close": 55.50,
53
   "volume": 30000
54
}
55
]
56
57
# insert the data into our collection
58
59
collection.insert_many(data)

1
collection.find_one({'timestamp': datetime(2023, 9, 12, 15, 19, 48)})

1
pip install pymongoarrow

1
import pymongoarrow
2
import pandas as pd
3
4
# pymongoarrow.monkey module provided an interface to patch pymongo, in place, and add pymongoarrow's functionality directly to collection instance. 
5
6
from pymongoarrow.monkey import patch_all
7
patch_all()
8
9
# Let's use the pymongoarrow's find_pandas_all() function to read MongoDB query result sets into 
10
11
df = collection.find_pandas_all({})

1
print(df)
2
3
print(type(df))

1
df = df.sort_values('timestamp')
2
3
df['daily_pct_change'] = df['close'].pct_change() * 100
4
5
# print the dataframe to see the modified data
6
print(df)

1
from pymongoarrow.api import write
2
3
coll = db.my_new_collection
4
5
# write data from pandas into MongoDB collection called 'coll'
6
write(coll, df)
7
8
# Now, let's verify that the modified data has been written into our collection
9
10
print(coll.find_one({}))