chibicitiberiu/collector
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Added stocks plugin

Browse Source
This commit is contained in:
Tiberiu Chibici
2020-04-17 19:16:58 +03:00
parent b7c1a3f6f4
commit daa8e5316c
13 changed files with 180 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
collector.py
View File

@@ -1,30 +1,38 @@
import config
import database
import signal
from apscheduler.schedulers.blocking import BlockingScheduler
from threading import Event
from plugins.system.cpu_plugin import CpuPlugin
from plugins.system.memory_plugin import MemoryPlugin
from plugins.system.disk_plugin import DiskPlugin
from plugins.system.disk_plugin import DiskIOPlugin, DiskUsagePlugin
from plugins.system.network_plugin import NetworkPlugin
from plugins.system.temperatures_plugin import TemperaturesPlugin
from plugins.system.ping_plugin import PingPlugin
from plugins.finance.stocks_plugin import StocksPlugin
from plugins.finance.robor_plugin import RoborPlugin
import logging
class Collector(object):
def __init__(self):
self.plugins = [
# system
CpuPlugin(),
MemoryPlugin(),
DiskPlugin(),
DiskUsagePlugin(),
DiskIOPlugin(),
NetworkPlugin(),
TemperaturesPlugin(),
PingPlugin(),
# finance
StocksPlugin(),
RoborPlugin()
]
self.event = Event()
self.scheduler = BlockingScheduler()
def collect_models(self):
models = []
@@ -32,28 +40,28 @@ class Collector(object):
models.extend(plugin.models)
return models
def initialize(self):
def schedule_plugins(self):
for plugin in self.plugins:
self.scheduler.add_job(plugin.execute, 'interval', seconds=plugin.get_interval())
def run(self):
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.INFO)
models = self.collect_models()
database.initialize_db()
database.DB.create_tables(models)
signal.signal(signal.SIGHUP, self.abort)
signal.signal(signal.SIGTERM, self.abort)
signal.signal(signal.SIGINT, self.abort)
def run(self):
self.initialize()
self.schedule_plugins()
# signal.signal(signal.SIGHUP, self.abort)
# signal.signal(signal.SIGTERM, self.abort)
# signal.signal(signal.SIGINT, self.abort)
print(f'Started.')
while not self.event.is_set():
for plugin in self.plugins:
# try:
plugin.execute()
# except BaseException as ex:
# print(ex)
self.event.wait(config.INTERVAL)
# TODO: calculate wait time based on execution time
try:
self.scheduler.start()
except (KeyboardInterrupt, SystemExit):
pass
print(f'Stopped.')

33
config.py
View File

@@ -1,5 +1,5 @@
# Collect interval in seconds
INTERVAL=30
DEFAULT_INTERVAL = 30
# Database URL, which defines connection settings.
#
@@ -26,19 +26,30 @@ DATABASE_URL = 'sqlite:///data.db'
### CPU
CPU_INTERVAL = DEFAULT_INTERVAL
# Store statistics per CPU, not only combined
CPU_PER_CPU = True
### Disk
# How often to poll space information, as a number of INTERVALs
DISK_SPACE_FREQUENCY = 10
DISK_USAGE_INTERVAL = DEFAULT_INTERVAL * 10
DISK_IO_INTERVAL = DEFAULT_INTERVAL
### Memory
MEMORY_INTERVAL = DEFAULT_INTERVAL
### Network
NETWORK_INTERVAL = DEFAULT_INTERVAL
### Temperatures
TEMPERATURE_INTERVAL = DEFAULT_INTERVAL
# If true, fahrenheit is used, otherwise celsius
TEMPERATURE_USE_FAHRENHEIT = False
### Ping
# Ping hosts
PING_INTERVAL = 5 * 60 # every 5 min
PING_HOSTS = [
'10.0.0.1',
'192.168.0.1',
@@ -47,11 +58,23 @@ PING_HOSTS = [
'bing.com'
]
# How often to send pings, as a number of INTERVALs
PING_FREQUENCY = 10
### Stocks
STOCKS_INTERVAL = 12 * 60 * 60 # updates daily
STOCKS_TICKERS = {
'MCRO.L' : 'Micro Focus International PLC',
'^GSPC' : 'S&P 500',
'^SPEUP' : 'S&P 350 Europe',
'^SPG1200' : 'S&P 1200 Global',
'^IXIC' : 'NASDAQ Composite',
'BTC-USD' : 'Bitcoin USD',
'ETH-USD' : 'Ethereum USD',
}
### ROBOR
# Romanian Interbank Offer Rate
ROBOR_INTERVAL = 12 * 60 * 60 # updates daily, every 12 hours should be fine
ROBOR_FIELDS = [
'ROBOR 6M'
]

4
plugins/finance/robor_plugin.py
View File

@@ -22,6 +22,9 @@ class RoborPlugin(Plugin):
def __init__(self):
self.__table = None
def get_interval(self):
return config.ROBOR_INTERVAL
def get_column_index(self, table, column_name):
header_row = table.find('tr')
for elem in header_row.iter('th'):
@@ -74,7 +77,6 @@ class RoborPlugin(Plugin):
entry.field = field
entry.value = value
entry.save()
print(model_to_dict(entry))

56
plugins/finance/stocks_plugin.py
View File

@@ -0,0 +1,56 @@
import datetime
from peewee import *
from playhouse.shortcuts import model_to_dict
import config
from database import BaseModel
from plugins.plugin import Plugin
import yfinance as yf
class Stocks(BaseModel):
date = DateTimeField(index=True, default=datetime.datetime.now(), null=False)
ticker = TextField(null=False)
label = TextField(null=False)
value_open = FloatField(null=False)
value_close = FloatField(null=False)
value_high = FloatField(null=False)
value_low = FloatField(null=False)
class StocksPlugin(Plugin):
models = [Stocks]
def get_interval(self):
return config.STOCKS_INTERVAL
def execute(self):
for ticker, label in config.STOCKS_TICKERS.items():
# Get last existing date
latest_date = Stocks.select(Stocks.date) \
.order_by(Stocks.date.desc()) \
.limit(1) \
.scalar()
try:
yfticker = yf.Ticker(ticker)
if latest_date is None:
data = yfticker.history(period='max')
else:
data = yfticker.history(start=latest_date + datetime.timedelta(seconds=1))
for row in data.itertuples():
entry = Stocks()
entry.date = row.Index.to_pydatetime()
entry.ticker = ticker
entry.label = label
entry.value_open = row.Open
entry.value_close = row.Close
entry.value_high = row.High
entry.value_low = row.Low
entry.save()
print(model_to_dict(entry))
except BaseException as e:
print(e)

0
plugins/finance_plugin.py
View File

4
plugins/plugin.py
View File

@@ -5,6 +5,10 @@ from typing import Tuple
class Plugin(ABC):
models = []
@abstractmethod
def get_interval(self) -> int:
pass
@abstractmethod
def execute(self) -> None:
pass

3
plugins/system/cpu_plugin.py
View File

@@ -27,6 +27,9 @@ class Cpu(BaseModel):
class CpuPlugin(Plugin):
models = [Cpu]
def get_interval(self):
return config.CPU_INTERVAL
def store(self, cpu, times, freq):
entry = Cpu()
entry.cpu = cpu

50
plugins/system/disk_plugin.py
View File

@@ -28,46 +28,52 @@ class DiskIO(BaseModel):
write_speed = FloatField(null=False)
class DiskPlugin(Plugin):
models = [DiskUsage, DiskIO]
class DiskUsagePlugin(Plugin):
models = [DiskUsage]
def get_interval(self):
return config.DISK_USAGE_INTERVAL
def execute(self):
for partition in psutil.disk_partitions():
usage = psutil.disk_usage(partition.mountpoint)
entry = DiskUsage()
entry.partition = partition.device
entry.mountpoint = partition.mountpoint
entry.total = usage.total
entry.used = usage.used
entry.free = usage.free
entry.save()
class DiskIOPlugin(Plugin):
models = [DiskIO]
def __init__(self):
self.__i = 0
self.__previous_io = {}
def get_interval(self):
return config.DISK_IO_INTERVAL
def store_io(self, disk, current):
previous = self.__previous_io.get(disk, current)
entry = DiskIO()
entry.disk = disk
entry.read_count = (current.read_count - previous.read_count) / config.INTERVAL
entry.write_count = (current.write_count - previous.write_count) / config.INTERVAL
entry.read_speed = (current.read_bytes - previous.read_bytes) / config.INTERVAL
entry.write_speed = (current.write_bytes - previous.write_bytes) / config.INTERVAL
entry.read_count = (current.read_count - previous.read_count) / self.get_interval()
entry.write_count = (current.write_count - previous.write_count) / self.get_interval()
entry.read_speed = (current.read_bytes - previous.read_bytes) / self.get_interval()
entry.write_speed = (current.write_bytes - previous.write_bytes) / self.get_interval()
entry.save()
self.__previous_io[disk] = current
def execute(self):
# Collect disk usage
if (self.__i % config.DISK_SPACE_FREQUENCY) == 0:
for partition in psutil.disk_partitions():
usage = psutil.disk_usage(partition.mountpoint)
entry = DiskUsage()
entry.partition = partition.device
entry.mountpoint = partition.mountpoint
entry.total = usage.total
entry.used = usage.used
entry.free = usage.free
entry.save()
# Collect IO
self.store_io(None, psutil.disk_io_counters(perdisk=False))
io_reads = psutil.disk_io_counters(perdisk=True)
for disk, current in io_reads.items():
self.store_io(disk, current)
self.__i += 1

3
plugins/system/memory_plugin.py
View File

@@ -26,6 +26,9 @@ class Memory(BaseModel):
class MemoryPlugin(Plugin):
models = [Memory]
def get_interval(self):
return config.MEMORY_INTERVAL
def execute(self):
vmem = psutil.virtual_memory()
swap = psutil.swap_memory()

11
plugins/system/network_plugin.py
View File

@@ -25,15 +25,18 @@ class NetworkPlugin(Plugin):
def __init__(self):
self.__previous_io = {}
def get_interval(self):
return config.NETWORK_INTERVAL
def store_io(self, nic, current):
previous = self.__previous_io.get(nic, current)
entry = NetworkIO()
entry.nic = nic
entry.packets_sent = (current.packets_sent - previous.packets_sent) / config.INTERVAL
entry.packets_recv = (current.packets_recv - previous.packets_recv) / config.INTERVAL
entry.bytes_sent = (current.bytes_sent - previous.bytes_sent) / config.INTERVAL
entry.bytes_recv = (current.bytes_recv - previous.bytes_recv) / config.INTERVAL
entry.packets_sent = (current.packets_sent - previous.packets_sent) / self.get_interval()
entry.packets_recv = (current.packets_recv - previous.packets_recv) / self.get_interval()
entry.bytes_sent = (current.bytes_sent - previous.bytes_sent) / self.get_interval()
entry.bytes_recv = (current.bytes_recv - previous.bytes_recv) / self.get_interval()
entry.save()
self.__previous_io[nic] = current

25
plugins/system/ping_plugin.py
View File

@@ -22,8 +22,10 @@ class PingPlugin(Plugin):
models = [Ping]
def __init__(self):
self.__timeout = config.INTERVAL // 3
self.__i = 0
self.__timeout = config.PING_INTERVAL // 3
def get_interval(self):
return config.PING_INTERVAL
async def do_ping(self, host):
command = ['ping', '-c', '1', '-W', str(self.__timeout), host]
@@ -47,13 +49,12 @@ class PingPlugin(Plugin):
await asyncio.gather(*[self.do_ping(host) for host in config.PING_HOSTS])
def execute(self):
if (self.__i % config.PING_FREQUENCY) == 0:
if getattr(asyncio, 'run', None) is not None:
# Python 3.7+
asyncio.run(self.execute_internal())
else:
loop = asyncio.get_event_loop()
loop.run_until_complete(self.execute_internal())
loop.close()
self.__i += 1
if getattr(asyncio, 'run', None) is not None:
# Python 3.7+
asyncio.run(self.execute_internal())
else:
loop = asyncio.get_event_loop()
loop.run_until_complete(self.execute_internal())
loop.close()

3
plugins/system/temperatures_plugin.py
View File

@@ -22,6 +22,9 @@ class Temperatures(BaseModel):
class TemperaturesPlugin(Plugin):
models = [Temperatures]
def get_interval(self):
return config.TEMPERATURE_INTERVAL
def execute(self):
for sensor, temps in psutil.sensors_temperatures(config.TEMPERATURE_USE_FAHRENHEIT).items():
for temp in temps:

8
requirements.txt
View File

@@ -1,8 +1,12 @@
apscheduler
peewee
# Used in most system plugins
psutil
# Finance plugins
# robor plugin
requests
lxml
lxml
# stocks plugin
yfinance