Skip to main content

A set of class tools to run the Funguana Pipeline. Many irrelelavent parts

Project description

Fungua Universe

Funbanner

This is a set of libraries to allow for the use of online live streaming systems within machine learning. They're largely separate from the services they're meant to support.

Installation

To install this code on your system you need to ensure you have pip installed on your system.

Run The Command

pip install funguauniverse

I'll need to heavily document this code. I'll go through a quick overview of what's included. The RL blog I crafted here, explains some of the uses of it.

For example, see how the background daemon works.

import threading
from crayons import yellow
from funguauniverse import PNode
from stochastic_agent.backtester.components.events import CheckNext
from stochastic_agent.backtester.components.parallel.policy import log_reward
from stochastic_agent.backtester.components.parallel.technical import reward_calculation


class RewardWorker(PNode):
    def __init__(self, event_queue, *args, **kwargs):
        super().__init__(event_queue, *args, **kwargs)
        # This is watching all futures to check if they're done
        self.future_list = []
        self.future_watcher = threading.Thread(target=self.signal_watch, daemon=True).start()

    def process(self):
        event = self.get_action_from_queue()
        if event is None:
            return


        episode = event.__dict__["episode"]
        # print(yellow(episode))
        # Get from episode data
        # Check for the changes using the exact same ideas from before.
        future = reward_calculation(episode, {})
        self.future_list.append(future)

    def create_next_event(self, episode=None):

        return CheckNext(episode=episode)

    def act_on_results(self, reward_obs):
        episode = reward_obs['episode']
        reward = reward_obs['reward']
        print(yellow(f"Reward: {reward}"))
        log_reward(episode, reward)
        # Do something about the action here
        # Create a new event
        next_event = self.create_next_event(episode)
        self.event_queue.put(next_event)

    def signal_watch(self):
        while True:
            # print(self.future_list)
            for index, future in enumerate(self.future_list):
                if future.done():
                    final = self.future_list.pop(index)
                    # print(final)
                    self.act_on_results(final.result())

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

funguauniverse-0.4.4.tar.gz (14.3 kB view hashes)

Uploaded Source

Built Distribution

funguauniverse-0.4.4-py3-none-any.whl (32.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page