Watcher nested aggregations with "direction"

You're right, Alex - I don't have an Execute API as I'm struggling with layout the aggregations in the watch.
WRT this search does not include any aggregation... That's what I'm struggling with - layout the aggregation definitions. Given a sample search hit:

  {
    "_index": "metrics-logstash-events-2018.05.15",
    "_type": "logs",
    "_id": "AWNhn5cNE5CHCIs4VPOu",
    "_score": 1.2332723,
    "_source": {
      "hostname": "idb-syslog-to-elk01",
      "@timestamp": "2018-05-15T02:27:18.923Z",
      "role": "idb-syslog-to-elk",
      "@version": "1",
      "message": "390d1450067e",
      "env": "dev",
      "events": {
        "rate_1m": 1359.9970463953305,
        "rate_15m": 1430.3099750447518,
        "count": 12333460050,
        "rate_5m": 1437.3180885655572
      },
      "direction": "in"
    }
  },

I'm trying to aggregate:

  1. Temporally (@timestamp)

  2. By host (hostname)

  3. by metric sum (events.count)

  4. By direction (direction)

I have hard time coming up with the definition of the aggregations.
In the condition portion of the watch I want to do mnemonically (as indicated above):

hosts[hostA].events.count[direction=IN] / hosts[hostA].events.count[direction=OUT]) * 100 > threshold

I'm trying to mimic a similar watch I did previously, but have hard time introducing another aggregation level (direction) and then coming up with the condition statement.
Is that something you can nudge me slightly in the right direction?
Thanks