/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.apex.examples.yahoofinance;

import org.apache.apex.malhar.lib.io.ConsoleOutputOperator;
import org.apache.apex.malhar.lib.math.RangeKeyVal;
import org.apache.apex.malhar.lib.math.SumKeyVal;
import org.apache.apex.malhar.lib.multiwindow.SimpleMovingAverage;
import org.apache.apex.malhar.lib.stream.ConsolidatorKeyVal;
import org.apache.apex.malhar.lib.util.BaseKeyValueOperator.DefaultPartitionCodec;
import org.apache.apex.malhar.lib.util.HighLow;
import org.apache.hadoop.conf.Configuration;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.Context.PortContext;
import com.datatorrent.api.DAG;
import com.datatorrent.api.Operator.InputPort;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.annotation.ApplicationAnnotation;

/**
 * Yahoo! Finance Application Example :<br>
 * Get Yahoo finance feed and calculate minute price range, minute volume,
 * simple moving average of 5 minutes. <br>
 * <br>
 * Functional Description : <br>
 * Application samples yahoo finance ticker every 200ms. All data points in one
 * second are streamed from input adapter. <br>
 * <br>
 *
 * Application calculates following Real Time Value(s):<br>
 * <ul>
 * <li>Quotes for IBM, Google, Apple, Yahoo stocks price/volume/time displayed
 * every second.</li>
 * <li>Charts for Stocks in terms for high/low price vs volume for last minute.</li>
 * <li>Simple moving average over last 5 minutes for IBM, Google, Apple, Yahoo
 * stocks.</li>
 * </ul>
 * <br>
 * <br>
 *
 * Custom Attribute : <br>
 * <ul>
 * <li>Application streaming window size(STREAMING_WINDOW_SIZE_MILLIS) = 1 sec,
 * since we are only interested in quotes every second.</li>
 * <li>Range/Minute Volume operator's window size(APPLICATION_WINDOW_COUNT) =
 * 60, aggregate over one minute.</li>
 * <li>Sum operator window length : 300, sliding average over last 5 minutes.</li>
 * </ul>
 * <br>
 *
 * Input Adapter : <br>
 * Stock Tick input operator get yahoo finance real time stock quotes data and
 * pushes application. <br>
 * <br>
 *
 * Output Adapter : <br>
 * Output values are written to console through ConsoleOutputOerator<br>
 * if you need to change write to HDFS,HTTP .. instead of console, <br>
 * Please refer to {@link org.apache.apex.malhar.lib.io.HttpOutputOperator} or
 * {@link org.apache.apex.malhar.lib.io.fs.HdfsOutputOperator}. <br>
 * <br>
 *
 * Run Sample Application : <br>
 * <p>
 * Running Java Test or Main app in IDE:
 *
 * <pre>
 * LocalMode.runApp(new Application(), 600000); // 10 min run
 * </pre>
 *
 * Run Success : <br>
 * For successful deployment and run, user should see following output on
 * console:
 *
 * <pre>
 *  Price SMA: AAPL=435.965
 * Price SMA: GOOG=877.0
 * QUOTE: {AABA=[26.37, 9760360, 4:00pm, null, null], IBM=[203.77, 2899698, 4:00pm, null, null], GOOG=[877.0, 2069614, 4:00pm, null, null], AAPL=[435.965, 10208099, 4:00pm, null, null]}
 * Price SMA: AABA=26.37
 * </pre>
 *
 * Scaling Options : <br>
 * <ul>
 * <li>Volume operator can be replicated using a {@link StatelessPartitioner}
 * on an operator.</li>
 * <li>Range value operator can replicated but using proper unifier
 * operator(read App Dev Guide).</li>
 * <li>Slinging window operator can be replicated with proper unifier operator.</li>
 * </ul>
 * <br>
 *
 * Application DAG : <br>
 * <img src="doc-files/Application.gif" width=600px > <br>
 * <br>
 *
 * Streaming Window Size : 1000 ms(1 Sec) <br>
 * Operator Details : <br>
 * <ul>
 * <li>
 * <p>
 * <b>The operator DailyVolume:</b> This operator reads from the input port,
 * which contains the incremental volume tuples from StockTickInput, and
 * aggregates the data to provide the cumulative volume. It just utilizes the
 * library class SumKeyVal<K,V> provided in math package. In this case,
 * SumKeyVal<String,Long>, where K is the stock symbol, V is the aggregated
 * volume, with cumulative set to true. (Otherwise if cumulative was set to
 * false, SumKeyVal would provide the sum for the application window.) The platform
 * provides a number of built-in operators for simple operations like this so
 * that application developers do not have to write them. More examples to
 * follow. This operator assumes that the application restarts before market
 * opens every day.
 * </p>
 * Class : {@link org.apache.apex.malhar.lib.math.SumKeyVal} <br>
 * Operator Application Window Count : 1 <br>
 * StateFull : Yes, volume gets aggregated every window count.</li>
 *
 * <li>
 * <p>
 * <b>The operator MinuteVolume:</b> This operator reads from the input port,
 * which contains the volume tuples from StockTickInput, and aggregates the data
 * to provide the sum of the volume within one minute. Like the operator
 * DailyVolume, this operator is also SumKeyVal<String,Long>, but with
 * cumulative set to false. Application Window is set to 1 minute. We will
 * explain how to set this later. <br>
 * Class : {@link org.apache.apex.malhar.lib.math.SumKeyVal} <br>
 * Operator App Window Count : 60 (1 Minute) <br>
 * StateFull : Yes, aggregate over last 60 windows.</li>
 *
 * <li>
 * <p>
 * <b>The operator Quote:</b> This operator has three input ports, which are
 * price (from StockTickInput), daily_vol (from Daily Volume), and time (from
 * StockTickInput). This operator just consolidates the three data and and emits
 * the consolidated data. It utilizes the class ConsolidatorKeyVal<K> from
 * stream package.<br>
 * Class : {@link org.apache.apex.malhar.lib.stream.ConsolidatorKeyVal} <br>
 * Operator App Window Count : 1 <br>
 * StateFull : No</li>
 *
 * <li>
 * <p>
 * <b>The operator Chart:</b> This operator is very similar to the operator
 * Quote, except that it takes inputs from High Low and Minute Vol and outputs
 * the consolidated tuples to the output port. <br>
 * Class : {@link org.apache.apex.malhar.lib.stream.ConsolidatorKeyVal} <br>
 * StateFull : No<br>
 * Operator App Window Count : 1</li>
 *
 *
 * <li>
 * <p>
 * <b>The operator PriceSMA:</b> SMA stands for - Simple Moving Average. It
 * reads from the input port, which contains the price tuples from
 * StockTickInput, and provides the moving average price of the stock. It
 * utilizes SimpleMovingAverage<String,Double>, which is provided in multiwindow
 * package. SimpleMovingAverage keeps track of the data of the previous N
 * application windows in a sliding manner. For each end window event, it
 * provides the average of the data in those application windows. <br>
 * Class : {@link org.apache.apex.malhar.lib.multiwindow.SimpleMovingAverage} <br>
 * StateFull : Yes, stores values across application window. <br>
 * Operator App Window : 1 <br>
 * Operator Sliding Window : 300 (5 mins).</li>
 *
 * <li>
 * <p>
 * <b>The operator Console: </b> This operator just outputs the input tuples to
 * the console (or stdout). In this example, there are four console operators,
 * which connect to the output of Quote, Chart, PriceSMA and VolumeSMA. In
 * practice, they should be replaced by operators which would do something about
 * the data, like drawing charts. </li>
 *
 * </ul>
 * <br>
 *
 * @since 0.3.2
 */
@ApplicationAnnotation(name = "YahooFinanceExample")
public class YahooFinanceApplication implements StreamingApplication
{
  protected int streamingWindowSizeMilliSeconds = 1000; // 1 second
  protected int appWindowCountMinute = 60;   // 1 minute
  protected int appWindowCountSMA = 300;  // 5 minute
  //protected String[] tickers = {"IBM", "GOOG", "AAPL", "AABA"};

  /**
   * Instantiate stock input operator for actual Yahoo finance ticks of symbol, last price, total daily volume, and last traded price.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @return StockTickInput instance.
   */
  public StockTickInput getStockTickInputOperator(String name, DAG dag)
  {
    StockTickInput oper = dag.addOperator(name, StockTickInput.class);
    oper.readIntervalMillis = 200;
    //oper.symbols = tickers;
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.math.SumKeyVal} operator
   * to sends total daily volume by adding volumes from each ticks.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @return SumKeyVal instance.
   */
  public SumKeyVal<String, Long> getDailyVolumeOperator(String name, DAG dag)
  {
    SumKeyVal<String, Long> oper = dag.addOperator(name, new SumKeyVal<String, Long>());
    oper.setType(Long.class);
    oper.setCumulative(true);
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.math.SumKeyVal} operator
   * Get aggregated volume of 1 minute and send at the end window of 1 minute.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @param appWindowCount Operator aggregate window size.
   * @return SumKeyVal instance.
   */
  public SumKeyVal<String, Long> getMinuteVolumeOperator(String name, DAG dag, int appWindowCount)
  {
    SumKeyVal<String, Long> oper = dag.addOperator(name, new SumKeyVal<String, Long>());
    oper.setType(Long.class);
    dag.getOperatorMeta(name).getAttributes().put(OperatorContext.APPLICATION_WINDOW_COUNT, appWindowCount);
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.math.RangeKeyVal} operator to get high/low
   * value for each key within given application window.
   * Get High-low range for 1 minute.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @param appWindowCount Operator aggregate window size.
   * @return RangeKeyVal instance.
   */
  public RangeKeyVal<String, Double> getHighLowOperator(String name, DAG dag, int appWindowCount)
  {
    RangeKeyVal<String, Double> oper = dag.addOperator(name, new RangeKeyVal<String, Double>());
    dag.getOperatorMeta(name).getAttributes().put(OperatorContext.APPLICATION_WINDOW_COUNT, appWindowCount);
    oper.setType(Double.class);
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.stream.ConsolidatorKeyVal} to send
   * Quote (Merge price, daily volume, time)
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @return ConsolidatorKeyVal instance.
   */
  public ConsolidatorKeyVal<String,Double,Long,String,?,?> getQuoteOperator(String name, DAG dag)
  {
    ConsolidatorKeyVal<String,Double,Long,String,?,?> oper = dag.addOperator(name, new ConsolidatorKeyVal<String,Double,Long,String,Object,Object>());
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.stream.ConsolidatorKeyVal} to send
   * Chart (Merge minute volume and minute high-low)
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @return ConsolidatorKeyVal instance.
   */
  public ConsolidatorKeyVal<String,HighLow<Double>,Long,?,?,?> getChartOperator(String name, DAG dag)
  {
    ConsolidatorKeyVal<String,HighLow<Double>,Long,?,?,?> oper = dag.addOperator(name, new ConsolidatorKeyVal<String,HighLow<Double>,Long,Object,Object,Object>());
    return oper;
  }

  /**
   * Instantiate {@link org.apache.apex.malhar.lib.multiwindow.SimpleMovingAverage} to calculate moving average for price
   * over given window size. Sliding window size is 1.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @param appWindowCount Operator aggregate window size.
   * @return SimpleMovingAverage instance.
   */
  public SimpleMovingAverage<String, Double> getPriceSimpleMovingAverageOperator(String name, DAG dag, int appWindowCount)
  {
    SimpleMovingAverage<String, Double> oper = dag.addOperator(name, new SimpleMovingAverage<String, Double>());
    oper.setWindowSize(appWindowCount);
    oper.setType(Double.class);
    return oper;
  }

  /**
   * Get console for output operator.
   * @param name  Operator name
   * @param dag   Application DAG graph.
   * @return input port for console output.
   */
  public InputPort<Object> getConsole(String name, /*String nodeName,*/ DAG dag, String prefix)
  {
    // hack to output to HTTP based on actual environment
        /*
     String serverAddr = System.getenv("MALHAR_AJAXSERVER_ADDRESS");
     if (serverAddr != null) {
     HttpOutputOperator<Object> oper = dag.addOperator(name, new HttpOutputOperator<Object>());
     oper.setResourceURL(URI.create("http://" + serverAddr + "/channel/" + nodeName));
     return oper.input;
     }
     */

    ConsoleOutputOperator oper = dag.addOperator(name, ConsoleOutputOperator.class);
    oper.setStringFormat(prefix + ": %s");
    return oper.input;
  }

  /**
   * Populate Yahoo Finance Example Application DAG.
   */
  @SuppressWarnings("unchecked")
  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {

    dag.getAttributes().put(DAG.STREAMING_WINDOW_SIZE_MILLIS, streamingWindowSizeMilliSeconds);

    StockTickInput tick = getStockTickInputOperator("StockTickInput", dag);
    SumKeyVal<String, Long> dailyVolume = getDailyVolumeOperator("DailyVolume", dag);
    ConsolidatorKeyVal<String,Double,Long,String,?,?> quoteOperator = getQuoteOperator("Quote", dag);

    RangeKeyVal<String, Double> highlow = getHighLowOperator("HighLow", dag, appWindowCountMinute);
    SumKeyVal<String, Long> minuteVolume = getMinuteVolumeOperator("MinuteVolume", dag, appWindowCountMinute);
    ConsolidatorKeyVal<String,HighLow<Double>,Long,?,?,?> chartOperator = getChartOperator("Chart", dag);

    SimpleMovingAverage<String, Double> priceSMA = getPriceSimpleMovingAverageOperator("PriceSMA", dag, appWindowCountSMA);
    DefaultPartitionCodec<String, Double> codec = new DefaultPartitionCodec<String, Double>();
    dag.setInputPortAttribute(highlow.data, PortContext.STREAM_CODEC, codec);
    dag.setInputPortAttribute(priceSMA.data, PortContext.STREAM_CODEC, codec);
    dag.addStream("price", tick.price, quoteOperator.in1, highlow.data, priceSMA.data);
    dag.addStream("vol", tick.volume, dailyVolume.data, minuteVolume.data);
    dag.addStream("time", tick.time, quoteOperator.in3);
    dag.addStream("daily_vol", dailyVolume.sum, quoteOperator.in2);

    dag.addStream("quote_data", quoteOperator.out, getConsole("quoteConsole", dag, "QUOTE"));

    dag.addStream("high_low", highlow.range, chartOperator.in1);
    dag.addStream("vol_1min", minuteVolume.sum, chartOperator.in2);
    dag.addStream("chart_data", chartOperator.out, getConsole("chartConsole", dag, "CHART"));

    dag.addStream("sma_price", priceSMA.doubleSMA, getConsole("priceSMAConsole", dag, "Price SMA"));

  }

}