How to use _multiprocess method in Molotov

Best Python code snippet using molotov_python

mfx_publisher.py

Source: mfx_publisher.py Github

copy

Full Screen

1import sys2import zmq3import json4from time import sleep5class MFXPublisher(object):6 """7 Publishes real-time market data to client strategies 8 """9 def __init__(self,10 _host='127.0.0.1', 11 _protocol='tcp', 12 _pub_port=42069,13 _sleep_delay=1,14 _multiprocess=True): 15 # ZeroMQ Host16 self._host = _host17 18 # Connection Protocol19 self._protocol = _protocol20 21 # TCP Connection URL Template22 self._url = self._protocol + ":/​/​" + self._host + ":"23 # SUB Socket Port24 self._pub_port = _pub_port25 self._multiprocess = _multiprocess26 27 if self._multiprocess == False:28 self.create_context()29 print("[SUCCESS] MetaFarmerX Market Feeds Live")30 # set the frequency of data updates (s)31 self._sleep_delay = _sleep_delay 32 # Market Data Client Configuration33 self._PUBLISH_MARKET_DATA = True34 # Run MetaFarmerX35 self._RUN = True36 ##########################################################################37 def create_context(self):38 """39 Creates ZeroMQ Context40 """41 # ZeroMQ Context42 self._zmq_context = zmq.Context()43 # Create Sockets44 self._pub_socket = self._zmq_context.socket(zmq.PUB)45 self._pub_socket.setsockopt(zmq.SNDHWM, 1)46 self._pub_socket.setsockopt(zmq.LINGER, 0)47 # Connect PUB Socket to send market data48 self._pub_socket.bind(self._url + str(self._pub_port)) 49 ########################################################################## 50 def set_delay(self, _frequency):51 """52 Set new frequency of market updates53 """54 self._sleep_delay = _frequency55 56 ########################################################################## 57 def remote_send(self, _socket, _data):58 """59 Send string on socket60 """61 try:62 _socket.send_json(_data)63 except zmq.error.Again:64 print("\nResource timeout... trying again")65 sleep(self._sleep_delay)66 67 ########################################################################## 68 def update_data(self):69 """70 Retreive market data71 """72 self.fetch_data()73 self.transform_data()74 75 ##########################################################################76 def on_tick(self):77 """78 Publish market data on PUB socket79 """80 if self._PUBLISH_MARKET_DATA:81 # update market data82 self.update_data()83 # send data to client through PUB socket84 self.remote_send(self._pub_socket, self.tranformed_data)85 ##########################################################################86 def run(self):87 """88 Run MetaFarmerX89 """90 if self._multiprocess:91 self.create_context()92 print("[SUCCESS] MetaFarmerX Market Feeds Live")93 while self._RUN:94 # blocking delay95 sleep(self._sleep_delay)...

Full Screen

Full Screen

evaluation.py

Source: evaluation.py Github

copy

Full Screen

1"""The genetic operation of evaluation.2This module defines the a basic form of the evaluation phase of bingo3evolutionary algorithms.4"""5from multiprocessing import Pool6class Evaluation:7 """Base phase for calculating fitness of a population.8 A base class for the fitness evaluation of populations of genetic9 individuals (list of chromosomes) in bingo. All individuals in the10 population are evaluated with a fitness function unless their fitness has11 already been set.12 Parameters13 ----------14 fitness_function : FitnessFunction15 The function class that is used to calculate fitnesses of individuals16 in the population.17 redundant : bool18 Whether to re-evaluate individuals that have been evaluated previously.19 Default False.20 multiprocess : int or bool21 Number of processes to use in parallel evaluation22 or False for serial evaluation.23 Default False.24 Attributes25 ----------26 fitness_function : FitnessFunction27 The function class that is used to calculate fitnesses of individuals28 in the population.29 eval_count : int30 the number of fitness function evaluations that have occurred31 """32 def __init__(self, fitness_function, redundant=False, multiprocess=False):33 self.fitness_function = fitness_function34 self._redundant = redundant35 self._multiprocess = multiprocess36 @property37 def eval_count(self):38 """int : the number of evaluations that have been performed"""39 return self.fitness_function.eval_count40 @eval_count.setter41 def eval_count(self, value):42 self.fitness_function.eval_count = value43 def __call__(self, population):44 """Evaluates the fitness of an individual45 Parameters46 ----------47 population : list of chromosomes48 population for which fitness should be calculated49 """50 if self._multiprocess:51 self._multiprocess_eval(population)52 else:53 self._serial_eval(population)54 def _serial_eval(self, population):55 for indv in population:56 if self._redundant or not indv.fit_set:57 indv.fitness = self.fitness_function(indv)58 def _multiprocess_eval(self, population):59 num_procs = self._multiprocess if isinstance(self._multiprocess, int) \60 else None61 with Pool(processes=num_procs) as pool:62 results = []63 for i, indv in enumerate(population):64 if self._redundant or not indv.fit_set:65 results.append(66 pool.apply_async(_fitness_job,67 (indv, self.fitness_function, i)))68 for res in results:69 indv, i = res.get()70 population[i] = indv71def _fitness_job(individual, fitness_function, population_index):72 individual.fitness = fitness_function(individual)...

Full Screen

Full Screen

Blogs

Check out the latest blogs from LambdaTest on this topic:

Why Agile Teams Have to Understand How to Analyze and Make adjustments

How do we acquire knowledge? This is one of the seemingly basic but critical questions you and your team members must ask and consider. We are experts; therefore, we understand why we study and what we should learn. However, many of us do not give enough thought to how we learn.

How to increase and maintain team motivation

The best agile teams are built from people who work together as one unit, where each team member has both the technical and the personal skills to allow the team to become self-organized, cross-functional, and self-motivated. These are all big words that I hear in almost every agile project. Still, the criteria to make a fantastic agile team are practically impossible to achieve without one major factor: motivation towards a common goal.

Dec’22 Updates: The All-New LT Browser 2.0, XCUI App Automation with HyperExecute, And More!

Greetings folks! With the new year finally upon us, we’re excited to announce a collection of brand-new product updates. At LambdaTest, we strive to provide you with a comprehensive test orchestration and execution platform to ensure the ultimate web and mobile experience.

How To Get Started With Cypress Debugging

One of the most important tasks of a software developer is not just writing code fast; it is the ability to find what causes errors and bugs whenever you encounter one and the ability to solve them quickly.

How to Recognize and Hire Top QA / DevOps Engineers

With the rising demand for new services and technologies in the IT, manufacturing, healthcare, and financial sector, QA/ DevOps engineering has become the most important part of software companies. Below is a list of some characteristics to look for when interviewing a potential candidate.

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Molotov automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful