Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- class TradingStrategyExecutor(object):
- """Executes a trading strategy, its computations. Computational resources."""
- def ExecuteStrategy(self, strategy, market_data, session):
- """Executes given strategy.
- Args:
- strategy: TradingStrategy.
- other: see TradingStrategy.Evaluate.
- """
- @classmethod
- def GetExecutor(cls):
- """Provides a suitable strategy executor for given strategy computation.
- Chooses between in-process and Hadoop executor, depending on estimated
- amount of computation and available machine resources.
- Returns:
- TradingStrategyExecutor. One of predefined classes.
- """
- class TradingStrategyInProcessExecutor(TradingStrategyExecutor):
- """Executes trading strategy computation in-process, on local machine."""
- def ExecuteStrategy(self, strategy, market_data, session):
- """See base class."""
- class TradingStrategyHadoopExecutor(TradingStrategyExecutor):
- """Executes trading strategy computation in a Hadoop cluster, distributed."""
- def __init__(self, hadoop_manager):
- """Constructor.
- Args:
- hadoop_manager: HadoopManager. For delegating computation to Hadoop.
- """
- def ExecuteStrategy(self, strategy, market_data, session):
- """See base class."""
- def _CreateJobConfiguration(self, strategy, market_data, session):
- """Creates a hadoop job description for given strategy execution.
- Returns:
- hadoop.Configuration. Hadoop job configuration describing given strategy
- execution, suitable for execution in a Hadoop cluster.
- """
- class HadoopManager(object):
- """Manager of Hadoop jobs. Starts / stops / monitors jobs.
- Interface to a Hadoop cluster.
- """
- def CreateJob(self, configuration):
- """Creates a new job from job configuration.
- Args:
- configuration: Configuration. Job to run.
- Returns:
- Job.
- """
- def GetActiveJobs(self):
- """Returns jobs that have not completed yet.
- Returns:
- list of Job.
- """
- def OnJobCompletion(self, callback):
- """Registers a callback to be called when a job completes (of fails).
- Args:
- callback: callable(Job).
- """
- class HadoopJob(object):
- """An executable task (a piece of computation) running in a Hadoop cluster."""
- def Run(self, wait_for_completion=True):
- """Runs the job.
- Args:
- wait_for_completion: bool. Whether to block until the job completes.
- """
- @property
- def status(self):
- """Job execution status."""
- def GetCounters(self, counters=()):
- """Returns job-specific counters.
- Args:
- counters: list of str. Counters to return. Defaults to all.
- Returns:
- dict, str -> Counter.
- """
- class MarketDataSlicer(object):
- """Generates slices of market data from complete market data.
- The slices are typically of short and equal duration.
- For distributing market data processing into independent tasks,
- over disjoint time slices.
- """
- def __init__(self, market_data):
- """Constructor.
- Args:
- market_data: MarketData. Complete market data to generate slices from.
- """
- def IterMarketDataSlices(self, num_slices=None, slice_duration=None):
- """Generates slices of market data.
- Args:
- num_slices: int. Desired number of equi-duration slices.
- slice_duration: datetime.timedelta. Desired duration of resulting slices.
- num_slices and slice_duration are mutually exclusive.
- Yields:
- MarketData. Slices.
- """
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement