How to I create a ticking time series with Reactive Extensions for Python (RxPY)?
Question: ## My Setup:

I have a list of stock price data in (time, price) tuples:

    
    
    from datetime import datetime
    prices = [(datetime(2015, 1, 9), 101.9), (datetime(2015, 1, 12), 101.5), (datetime(2015, 1, 13), 101.7)]
    

which I want to make into an [RxPY](https://github.com/ReactiveX/RxPY)
`Observable` so that I may backtest a trading strategy tick-by-tick:

    
    
    def myStrategy(date, price):   # SELL if date is a Monday and price is above 101.6
        strategy = 'SELL' if date.weekday() and price > 101.6 else 'BUY'
        print 'date=%s price=%s strategy=%s' % (date, price, strategy)
    

I wish to start backtesting from 12 Jan 2015 so I assume I must use the
following scheduler:

    
    
    from rx.concurrency import HistoricalScheduler
    scheduler = HistoricalScheduler(datetime(2015, 1, 12))
    

To run my backtest I do:

    
    
    from rx import Observable
    observable = Observable.from_iterable(prices, scheduler=scheduler).timestamp()
    observable.subscribe(lambda price: myStrategy(price.timestamp, price.value))
    scheduler.start()
    

## Problem:

I expected to see:

    
    
    date=2015-01-12 00:00:00 price=101.5 strategy=BUY
    date=2015-01-13 00:00:00 price=101.7 strategy=SELL
    

but I got

    
    
    date=2015-12-20 08:43:45.882000 price=(datetime.datetime(2015, 1, 9, 0, 0), 101.9) strategy=SELL
    date=2015-12-20 08:43:45.882000 price=(datetime.datetime(2015, 1, 12, 0, 0), 101.5) strategy=SELL
    date=2015-12-20 08:43:45.882000 price=(datetime.datetime(2015, 1, 13, 0, 0), 101.7) strategy=SELL
    

The problems are that:

  * The timestamps are wrong: I get today's date of `2015-12-20 08:43:45.882000` instead of the historic dates (e.g. `datetime(2015, 1, 12)`)
  * The price still contains the time component
  * The scheduler did not start on 12 Jan 2015 as I requested, since I see that the data point from 9 Jan 2015 was still used.

I also tried using `scheduler.now()`:

    
    
    observable.subscribe(lambda price: myStrategy(scheduler.now(), price.value))
    

but then the date was stuck on `date=2015-01-12 00:00:00` for some reason:

    
    
    date=2015-01-12 00:00:00 price=(datetime.datetime(2015, 1, 9, 0, 0), 101.9) strategy=BUY
    date=2015-01-12 00:00:00 price=(datetime.datetime(2015, 1, 12, 0, 0), 101.5) strategy=BUY
    date=2015-01-12 00:00:00 price=(datetime.datetime(2015, 1, 13, 0, 0), 101.7) strategy=BUY
    

How do I fix the above and obtain the result I originally expected?


Answer: Also very new to rx,

  * I think the timestamp() needs the scheduler parameter. Otherwise it works on some default scheduler as per rx docs.
  * You are passing the entire tuple (date,price) as price to myStrategy() thats why it is printing the date.

"timestamp by default operates on the timeout Scheduler, but also has a
variant that allows you to specify the Scheduler by passing it in as a
parameter." <http://reactivex.io/documentation/operators/timestamp.html>

There is documentation only for rxjs , but the beauty of rx is that everything
follows.

Please see if this can work for you.

    
    
        observable = Observable.from_iterable(prices,scheduler=scheduler).timestamp(scheduler=scheduler)
        observable.subscribe(lambda price: myStrategy(price.timestamp, price.value[1]))
        scheduler.start()
    

