cache.java

package daytrader;

import com.ib.client.Contract;
import daytrader.cacheTypes.Request;
import daytrader.datamodel.BarPoint;
import daytrader.datamodel.BarPointGraph;
import daytrader.hibernate.BarEntity;
import daytrader.hibernate.ContractEntity;
import daytrader.hibernate.RequestEntity;
import daytrader.hibernate.SessionHelper;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.Transaction;

import javax.persistence.NoResultException;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Abstracts the interactions between the cache and the client
 */
public class EntityHandler {

    private final ReentrantLock lock;
    private final Logger logger = LogManager.getLogger();

    /**
     * Constructor
     */
    public EntityHandler() {
        lock = new ReentrantLock();
    }

    /**
     * Insert contract long.
     *
     * @param contract the contract
     * @return the long
     */
    public long insertContract(Contract contract) {
        ContractEntity entityTest = getContract(contract.symbol(), contract.exchange());
        if (entityTest != null) {
            return entityTest.getId();
        }

        boolean success = true;
        long contractId = 0;
        try {
            contractId = doInTransaction(session -> {
                ContractEntity contractEntity = new ContractEntity(contract);
                return (long) session.save(contractEntity);
            });
        } catch (Exception e) {
            logger.error(e);
            success = false;
        }

        if (success) {
            logger.info(String.format("Inserted contract: %s:%s", contract.symbol(), contract.exchange()));
        }

        return contractId;
    }

    /**
     * Insert contract long.
     *
     * @param symbol         the symbol
     * @param secType        the sec type
     * @param strike         the strike
     * @param exchange       the exchange
     * @param includeExpired the include expired
     * @return the long
     */
    public long insertContract(String symbol, String secType, double strike, String exchange, boolean includeExpired) {
        ContractEntity entityTest = getContract(symbol, exchange);
        if (entityTest != null) {
            return entityTest.getId();
        }

        boolean success = true;
        long contractId = 0;
        try {
            contractId = doInTransaction(session -> {
                ContractEntity contractEntity = new ContractEntity(symbol, secType, strike, exchange, includeExpired);
                return (long) session.save(contractEntity);
            });
        } catch (Exception e) {
            logger.error(e);
            success = false;
        }

        if (success) {
            logger.info(String.format("Inserted contract: %s:%s", symbol, exchange));
        }

        return contractId;
    }

    /**
     * Insert request long.
     *
     * @param request the request
     * @return the long
     */
    public long insertRequest(Request request) {
        RequestEntity entityTest = getRequest(request);
        if (entityTest != null) {
            return entityTest.getId();
        }

        if (getContract(request.getContract().symbol(), request.getContract().exchange()) == null) {
            insertContract(request.getContract());
        }

        long requestId = 0;
        ContractEntity contractEntity = getContract(request.getContract().symbol(), request.getContract().exchange());
        boolean success = true;
        try {
            requestId = doInTransaction(session -> {
                int count = 1;
                RequestEntity requestEntity = new RequestEntity(contractEntity, request.getTask().getStartDateTime(), request.getTask().getEndDateTime());
                session.save(requestEntity);
                for (ListIterator<BarPoint> iterator = request.getPoints().stream().collect(Collectors.toList()).listIterator(); iterator.hasNext(); ) {
                    BarPoint barPoint = iterator.next();
                    BarPoint next = null;
                    if (iterator.hasNext()) {
                        next = iterator.next();
                        iterator.previous();
                    }

                    if (next != null) {
                        if (!(barPoint.getOpen() == next.getOpen()
                                && barPoint.getClose() == next.getClose()
                                && barPoint.getHigh() == next.getHigh()
                                && barPoint.getLow() == next.getLow())) {
                            BarEntity barEntity = new BarEntity(barPoint.getDateTime(), barPoint.getOpen(), barPoint.getHigh(), barPoint.getLow(), barPoint.getClose(), barPoint.getWAP(), count, requestEntity);
                            session.save(barEntity);
                        }
                    } else {
                        BarEntity barEntity = new BarEntity(barPoint.getDateTime(), barPoint.getOpen(), barPoint.getHigh(), barPoint.getLow(), barPoint.getClose(), barPoint.getWAP(), count, requestEntity);
                        session.save(barEntity);
                    }

                    count++;
                }
                return requestEntity.getId();
            });
        } catch (Exception e) {
            logger.error(e);
            success = false;
        }

        if (success) {
            logger.info(String.format("Inserting request: %s:%s for %s-%s on %s",
                    request.getContract().symbol(),
                    request.getContract().exchange(),
                    request.getTask().getStartDateTime().toLocalTime(),
                    request.getTask().getEndDateTime().toLocalTime(),
                    request.getTask().getEndDateTime().toLocalDate()));
        }

        return requestId;
    }

    /**
     * Gets points by request.
     *
     * @param request the request
     * @return the points by request
     */
    public BarPointGraph getPointsByRequest(Request request) {
        RequestEntity requestEntity = getRequest(request);
        BarPointGraph points = new BarPointGraph();
        List<BarEntity> pointList = doInTransaction(session ->
                session.createQuery("FROM BarEntity e where e.requestEntity = :entity", BarEntity.class)
                        .setParameter("entity", requestEntity)
                        .list()
        );

        int count = 1;
        for (int i = 0, pointListSize = pointList.size(); i < pointListSize; i++) {
            BarEntity bar = pointList.get(i);
            while (bar.getRequestCount() > count) {
                BarPoint point = new BarPoint(bar);
                int diff = (int) (bar.getRequestCount() - count);
                point.setDateTime(bar.getDateTime().minusSeconds(diff * 5));
                points.add(point);
                count++;
            }
            points.add(bar);
        }
        return points;
    }

    /**
     * Gets contract.
     *
     * @param symbol   the symbol
     * @param exchange the exchange
     * @return the contract
     */
    public ContractEntity getContract(String symbol, String exchange) {
        ContractEntity contractEntity = null;

        try {
            contractEntity = doInTransaction(session ->
                    session.createQuery("SELECT e FROM ContractEntity e where e.symbol = :symbol AND e.exchange = :exchange", ContractEntity.class)
                            .setParameter("symbol", symbol)
                            .setParameter("exchange", exchange)
                            .getSingleResult());
        } catch (Exception ignored) {
        }

        return contractEntity;
    }

    /**
     * Gets request.
     *
     * @param request the request
     * @return the request
     */
    public RequestEntity getRequest(Request request) {
        RequestEntity requestEntity = null;

        try {
            requestEntity = doInTransaction(session ->
                    session.createQuery("SELECT e FROM RequestEntity e " +
                            "where e.contractEntity.symbol = :symbol AND e.contractEntity.exchange = :exchange AND e.startDate = :startDate AND e.endDate = :endDate", RequestEntity.class)
                            .setParameter("symbol", request.getContract().symbol())
                            .setParameter("exchange", request.getContract().exchange())
                            .setParameter("startDate", request.getTask().getStartDateTime().toString())
                            .setParameter("endDate", request.getTask().getEndDateTime().toString()).getSingleResult());
        } catch (NoResultException ignored) {
        }

        return requestEntity;
    }

    /**
     * The interface Hibernate transaction function.
     *
     * @param <T> the type parameter
     */
    @FunctionalInterface
    protected interface HibernateTransactionFunction<T> extends Function<Session, T> {
        /**
         * Before transaction completion.
         */
        default void beforeTransactionCompletion() {

        }

        /**
         * After transaction completion.
         */
        default void afterTransactionCompletion() {

        }
    }

    /**
     * Do in transaction t.
     *
     * @param <T>      the type parameter
     * @param callable the callable
     * @return the t
     */
    protected <T> T doInTransaction(HibernateTransactionFunction<T> callable) {
        T result = null;
        SessionHelper helper = new SessionHelper();
        SessionFactory sessionFactory = helper.getSessionFactory();
        lock.lock();
        Transaction transaction = null;

        try (Session session = sessionFactory.openSession()) {
            callable.beforeTransactionCompletion();
            transaction = session.beginTransaction();

            result = callable.apply(session);
            transaction.commit();
        } catch (NoResultException e) {
            throw e;
        } catch (RuntimeException e) {
            if (transaction != null && transaction.isActive()) transaction.rollback();
            throw e;
        } finally {
            callable.afterTransactionCompletion();
            lock.unlock();
        }

        return result;
    }
}